Add Packet Drop Events to RPS Scenario/Keyword (#2899)

src/core/worker.c

@@ -644,8 +644,6 @@ QuicWorkerLoop(
             1);
     }
 
-    Context->Ready = FALSE;
-
     //
     // Opportunistically try to snap-shot performance counters and do some
     // validation.
@@ -733,7 +731,8 @@ CXPLAT_THREAD_CALLBACK(QuicWorkerThread, Context)
 
     uint64_t TimeNow = CxPlatTimeUs64();
     while (QuicWorkerLoop(EC, &TimeNow, ThreadID)) {
-        if (!EC->Ready) {
+        BOOLEAN Ready = InterlockedFetchAndClearBoolean(&EC->Ready);
+        if (!Ready) {
             if (EC->NextTimeUs == UINT64_MAX) {
                 CxPlatEventWaitForever(Worker->Ready);
                 TimeNow = CxPlatTimeUs64();

src/inc/quic_platform.h

@@ -450,7 +450,7 @@ typedef struct CXPLAT_EXECUTION_CONTEXT {
     void* CxPlatContext;
     CXPLAT_EXECUTION_FN Callback;
     uint64_t NextTimeUs;
-    BOOLEAN Ready;
+    volatile BOOLEAN Ready;
 
 } CXPLAT_EXECUTION_CONTEXT;
 

src/inc/quic_platform_posix.h

@@ -175,6 +175,15 @@ InterlockedCompareExchange64(
     return __sync_val_compare_and_swap(Destination, Comperand, ExChange);
 }
 
+inline
+BOOLEAN
+InterlockedFetchAndClearBoolean(
+    _Inout_ _Interlocked_operand_ BOOLEAN volatile *Target
+    )
+{
+    return __sync_fetch_and_and(Target, 0);
+}
+
 inline
 void*
 InterlockedFetchAndClearPointer(

src/inc/quic_platform_winkernel.h

@@ -112,6 +112,15 @@ InterlockedFetchAndClearPointer(
     return InterlockedExchangePointer(Target, NULL);
 }
 
+inline
+BOOLEAN
+InterlockedFetchAndClearBoolean(
+    _Inout_ _Interlocked_operand_ BOOLEAN volatile *Target
+    )
+{
+    return (BOOLEAN)InterlockedAnd8((char*)Target, 0);
+}
+
 //
 // Static Analysis Interfaces
 //

src/inc/quic_platform_winuser.h

@@ -220,6 +220,15 @@ InterlockedFetchAndClearPointer(
     return InterlockedExchangePointer(Target, NULL);
 }
 
+inline
+BOOLEAN
+InterlockedFetchAndClearBoolean(
+    _Inout_ _Interlocked_operand_ BOOLEAN volatile *Target
+    )
+{
+    return (BOOLEAN)InterlockedAnd8((char*)Target, 0);
+}
+
 //
 // CloseHandle has an incorrect SAL annotation, so call through a wrapper.
 //

src/manifest/MsQuicEtw.man

@@ -2631,7 +2631,7 @@
               value="5153"
               />
           <event
-              keywords="ut:Connection ut:Packet"
+              keywords="ut:Connection ut:Packet ut:RPS"
               level="win:Informational"
               message="$(string.Etw.ConnDropPacket)"
               opcode="Connection"
@@ -2640,7 +2640,7 @@
               value="5154"
               />
           <event
-              keywords="ut:Connection ut:Packet"
+              keywords="ut:Connection ut:Packet ut:RPS"
               level="win:Informational"
               message="$(string.Etw.ConnDropPacketEx)"
               opcode="Connection"
@@ -2740,7 +2740,7 @@
               />
           <event
               keywords="ut:Connection ut:Packet"
-              level="win:Verbose"
+              level="win:Informational"
               message="$(string.Etw.ConnPacketLost)"
               opcode="Connection"
               symbol="QuicConnPacketLost"
@@ -3137,7 +3137,7 @@
               value="7171"
               />
           <event
-              keywords="ut:Binding ut:Packet"
+              keywords="ut:Binding ut:Packet ut:RPS"
               level="win:Informational"
               message="$(string.Etw.BindingDropPacket)"
               opcode="Binding"
@@ -3146,7 +3146,7 @@
               value="7172"
               />
           <event
-              keywords="ut:Binding ut:Packet"
+              keywords="ut:Binding ut:Packet ut:RPS"
               level="win:Informational"
               message="$(string.Etw.BindingDropPacketEx)"
               opcode="Binding"

src/perf/lib/RpsClient.cpp

@@ -39,6 +39,7 @@ PrintHelp(
         "  -threads:<####>             The number of threads to use. Defaults and capped to number of cores\n"
         "  -affinitize:<0/1>           Affinitizes threads to a core. (def:0)\n"
         "  -sendbuf:<0/1>              Whether to use send buffering. (def:0)\n"
+        "  -stats:<0/1>                Indicates connection stats should be printed at the end of the run. (def:0)\n"
         "\n",
         RPS_MAX_CLIENT_PORT_COUNT,
         RPS_DEFAULT_RUN_TIME,
@@ -111,6 +112,7 @@ RpsClient::Init(
     TryGetValue(argc, argv, "requests", &RequestCount);
     TryGetValue(argc, argv, "request", &RequestLength);
     TryGetValue(argc, argv, "response", &ResponseLength);
+    TryGetValue(argc, argv, "stats", &PrintStats);
 
     const char* CibirBytes = nullptr;
     if (TryGetValue(argc, argv, "cibir", &CibirBytes)) {
@@ -436,6 +438,9 @@ RpsConnectionContext::ConnectionCallback(
         //WriteOutput("Connection died, 0x%x\n", Event->SHUTDOWN_INITIATED_BY_TRANSPORT.Status);
         break;
     case QUIC_CONNECTION_EVENT_SHUTDOWN_COMPLETE:
+        if (Client->PrintStats) {
+            QuicPrintConnectionStatistics(MsQuic, Handle);
+        }
         break;
     case QUIC_CONNECTION_EVENT_IDEAL_PROCESSOR_CHANGED:
         if ((uint32_t)Event->IDEAL_PROCESSOR_CHANGED.IdealProcessor >= Client->WorkerCount) {

src/perf/lib/RpsClient.h

@@ -178,6 +178,7 @@ public:
     UniquePtr<char[]> Target;
     uint8_t UseEncryption {TRUE};
     uint8_t SendInline {FALSE};
+    uint8_t PrintStats {FALSE};
     uint32_t RunTime {RPS_DEFAULT_RUN_TIME};
     uint32_t ConnectionCount {RPS_DEFAULT_CONNECTION_COUNT};
     uint32_t RequestCount {RPS_DEFAULT_CONNECTION_COUNT * 2};

src/platform/platform_worker.c

@@ -248,7 +248,8 @@ CxPlatRunExecutionContexts(
     while (*EC != NULL) {
         CXPLAT_EXECUTION_CONTEXT* Context =
             CXPLAT_CONTAINING_RECORD(*EC, CXPLAT_EXECUTION_CONTEXT, Entry);
-        if (Context->Ready || Context->NextTimeUs <= *TimeNow) {
+        BOOLEAN Ready = InterlockedFetchAndClearBoolean(&Context->Ready);
+        if (Ready || Context->NextTimeUs <= *TimeNow) {
             CXPLAT_SLIST_ENTRY* Next = Context->Entry.Next;
             DidWork = TRUE;
             if (!Context->Callback(Context->Context, TimeNow, Worker->ThreadId)) {