rts: Flush eventlog buffers from flushEventLogwip/T18043

As noted in #18043, flushTrace failed flush anything beyond the writer.
This means that a significant amount of data sitting in capability-local
event buffers may never get flushed, despite the users' pleads for us to
flush.

Fix this by making flushEventLog flush all of the event buffers before
flushing the writer.

Fixes #18043.

11 files changed, 75 insertions, 11 deletions

diff --git a/includes/RtsAPI.h b/includes/RtsAPI.h
index 36ab4d4b04..f88cd0614c 100644
--- a/includes/RtsAPI.h
+++ b/includes/RtsAPI.h
@@ -18,8 +18,6 @@ extern "C" {
 #include "HsFFI.h"
 #include "rts/Time.h"
 #include "rts/Types.h"
-#include "rts/EventLogWriter.h"
-
 
 /*
  * Running the scheduler
@@ -60,6 +58,9 @@ typedef struct CapabilityPublic_ {
     StgRegTable r;
 } CapabilityPublic;
 
+/* N.B. this needs the Capability declaration above. */
+#include "rts/EventLogWriter.h"
+
 /* ----------------------------------------------------------------------------
    RTS configuration settings, for passing to hs_init_ghc()
    ------------------------------------------------------------------------- */
diff --git a/includes/rts/EventLogWriter.h b/includes/rts/EventLogWriter.h
index 38a0c06086..73a2aec64c 100644
--- a/includes/rts/EventLogWriter.h
+++ b/includes/rts/EventLogWriter.h
@@ -68,3 +68,8 @@ bool startEventLogging(const EventLogWriter *writer);
  * Stop event logging and destroy the current EventLogWriter.
  */
 void endEventLogging(void);
+
+/*
+ * Flush the eventlog. cap can be NULL if one is not held.
+ */
+void flushEventLog(Capability **cap);
diff --git a/libraries/base/Debug/Trace.hs b/libraries/base/Debug/Trace.hs
index 453b5141b4..5a8b55126c 100644
--- a/libraries/base/Debug/Trace.hs
+++ b/libraries/base/Debug/Trace.hs
@@ -37,6 +37,7 @@ module Debug.Trace (
         -- $eventlog_tracing
         traceEvent,
         traceEventIO,
+        flushEventLog,
 
         -- * Execution phase markers
         -- $markers
@@ -319,3 +320,11 @@ traceMarkerIO :: String -> IO ()
 traceMarkerIO msg =
   GHC.Foreign.withCString utf8 msg $ \(Ptr p) -> IO $ \s ->
     case traceMarker# p s of s' -> (# s', () #)
+
+-- | Immediately flush the event log, if enabled.
+--
+-- @since 4.15.0.0
+flushEventLog :: IO ()
+flushEventLog = c_flushEventLog nullPtr
+
+foreign import ccall "flushEventLog" c_flushEventLog :: Ptr () -> IO ()
diff --git a/rts/Capability.c b/rts/Capability.c
index a655fc7b3f..dd24ce6681 100644
--- a/rts/Capability.c
+++ b/rts/Capability.c
@@ -23,6 +23,7 @@
 #include "Schedule.h"
 #include "Sparks.h"
 #include "Trace.h"
+#include "eventlog/EventLog.h" // for flushLocalEventsBuf
 #include "sm/GC.h" // for gcWorkerThread()
 #include "STM.h"
 #include "RtsUtils.h"
@@ -982,6 +983,10 @@ yieldCapability
                 debugTrace(DEBUG_nonmoving_gc, "Flushing update remembered set blocks...");
                 break;
 
+            case SYNC_FLUSH_EVENT_LOG:
+                flushLocalEventsBuf(cap);
+                break;
+
             default:
                 break;
             }
diff --git a/rts/Capability.h b/rts/Capability.h
index 8c5b1e814e..df486829ea 100644
--- a/rts/Capability.h
+++ b/rts/Capability.h
@@ -27,6 +27,7 @@
 
 #include "BeginPrivate.h"
 
+/* N.B. This must be consistent with CapabilityPublic in RtsAPI.h */
 struct Capability_ {
     // State required by the STG virtual machine when running Haskell
     // code.  During STG execution, the BaseReg register always points
@@ -267,7 +268,8 @@ typedef enum {
     SYNC_OTHER,
     SYNC_GC_SEQ,
     SYNC_GC_PAR,
-    SYNC_FLUSH_UPD_REM_SET
+    SYNC_FLUSH_UPD_REM_SET,
+    SYNC_FLUSH_EVENT_LOG
 } SyncType;
 
 //
diff --git a/rts/RtsSymbols.c b/rts/RtsSymbols.c
index e433d9d369..7cd62faabe 100644
--- a/rts/RtsSymbols.c
+++ b/rts/RtsSymbols.c
@@ -594,6 +594,7 @@
       SymI_HasProto(__word_encodeFloat)                                 \
       SymI_HasProto(stg_atomicallyzh)                                   \
       SymI_HasProto(barf)                                               \
+      SymI_HasProto(flushEventLog)                                      \
       SymI_HasProto(deRefStablePtr)                                     \
       SymI_HasProto(debugBelch)                                         \
       SymI_HasProto(errorBelch)                                         \
diff --git a/rts/Schedule.c b/rts/Schedule.c
index b7c1e7b0c0..a5bd11bf5e 100644
--- a/rts/Schedule.c
+++ b/rts/Schedule.c
@@ -2070,7 +2070,7 @@ forkProcess(HsStablePtr *entry
     stopTimer(); // See #4074
 
 #if defined(TRACING)
-    flushEventLog(); // so that child won't inherit dirty file buffers
+    flushAllCapsEventsBufs(); // so that child won't inherit dirty file buffers
 #endif
 
     pid = fork();
diff --git a/rts/Trace.c b/rts/Trace.c
index 6d77cc1254..2a229b5dc2 100644
--- a/rts/Trace.c
+++ b/rts/Trace.c
@@ -118,10 +118,10 @@ void resetTracing (void)
     restartEventLogging();
 }
 
-void flushTrace (void)
+void flushTrace ()
 {
     if (eventlog_enabled) {
-        flushEventLog();
+        flushEventLog(NULL);
     }
 }
 
diff --git a/rts/Trace.h b/rts/Trace.h
index 9c905af737..cad83363d3 100644
--- a/rts/Trace.h
+++ b/rts/Trace.h
@@ -319,7 +319,6 @@ void traceConcSweepEnd(void);
 void traceConcUpdRemSetFlush(Capability *cap);
 void traceNonmovingHeapCensus(uint32_t log_blk_size,
                               const struct NonmovingAllocCensus *census);
-
 void flushTrace(void);
 
 #else /* !TRACING */
diff --git a/rts/eventlog/EventLog.c b/rts/eventlog/EventLog.c
index 11e8a5e0b6..bce2fa493f 100644
--- a/rts/eventlog/EventLog.c
+++ b/rts/eventlog/EventLog.c
@@ -16,6 +16,7 @@
 #include "RtsUtils.h"
 #include "Stats.h"
 #include "EventLog.h"
+#include "Schedule.h"
 
 #include <string.h>
 #include <stdio.h>
@@ -270,8 +271,8 @@ stopEventLogWriter(void)
     }
 }
 
-void
-flushEventLog(void)
+static void
+flushEventLogWriter(void)
 {
     if (event_log_writer != NULL &&
             event_log_writer->flushEventLog != NULL) {
@@ -1484,7 +1485,7 @@ void printAndClearEventBuf (EventsBuf *ebuf)
                     "printAndClearEventLog: could not flush event log\n"
                 );
             resetEventsBuf(ebuf);
-            flushEventLog();
+            flushEventLogWriter();
             return;
         }
 
@@ -1566,6 +1567,40 @@ void postEventType(EventsBuf *eb, EventType *et)
     postInt32(eb, EVENT_ET_END);
 }
 
+void flushLocalEventsBuf(Capability *cap)
+{
+    EventsBuf *eb = &capEventBuf[cap->no];
+    printAndClearEventBuf(eb);
+}
+
+// Flush all capabilities' event buffers when we already hold all capabilities.
+// Used during forkProcess.
+void flushAllCapsEventsBufs()
+{
+    ACQUIRE_LOCK(&eventBufMutex);
+    printAndClearEventBuf(&eventBuf);
+    RELEASE_LOCK(&eventBufMutex);
+
+    for (unsigned int i=0; i < n_capabilities; i++) {
+        flushLocalEventsBuf(capabilities[i]);
+    }
+    flushEventLogWriter();
+}
+
+void flushEventLog(Capability **cap USED_IF_THREADS)
+{
+    ACQUIRE_LOCK(&eventBufMutex);
+    printAndClearEventBuf(&eventBuf);
+    RELEASE_LOCK(&eventBufMutex);
+
+#if defined(THREADED_RTS)
+    Task *task = getMyTask();
+    stopAllCapabilitiesWith(cap, task, SYNC_FLUSH_EVENT_LOG);
+    releaseAllCapabilities(n_capabilities, cap ? *cap : NULL, task);
+#endif
+    flushEventLogWriter();
+}
+
 #else
 
 enum EventLogStatus eventLogStatus(void)
@@ -1579,4 +1614,6 @@ bool startEventLogging(const EventLogWriter *writer STG_UNUSED) {
 
 void endEventLogging(void) {}
 
+void flushEventLog(Capability **cap STG_UNUSED) {}
+
 #endif /* TRACING */
diff --git a/rts/eventlog/EventLog.h b/rts/eventlog/EventLog.h
index eca76619cd..04ed1982a3 100644
--- a/rts/eventlog/EventLog.h
+++ b/rts/eventlog/EventLog.h
@@ -28,8 +28,10 @@ void initEventLogging(void);
 void restartEventLogging(void);
 void freeEventLogging(void);
 void abortEventLogging(void); // #4512 - after fork child needs to abort
-void flushEventLog(void);     // event log inherited from parent
 void moreCapEventBufs (uint32_t from, uint32_t to);
+void flushLocalEventsBuf(Capability *cap);
+void flushAllCapsEventsBufs(void);
+void flushAllEventsBufs(Capability *cap);
 
 /*
  * Post a scheduler event to the capability's event buffer (an event
@@ -175,6 +177,9 @@ void postNonmovingHeapCensus(int log_blk_size,
 
 #else /* !TRACING */
 
+INLINE_HEADER void flushLocalEventsBuf(Capability *cap STG_UNUSED)
+{ /* nothing */ }
+
 INLINE_HEADER void postSchedEvent (Capability *cap  STG_UNUSED,
                                    EventTypeNum tag STG_UNUSED,
                                    StgThreadID id   STG_UNUSED,