[EventPipe] Add Blocking EventPipe Mode

src/coreclr/inc/clrconfigvalues.h

@@ -611,6 +611,7 @@ RETAIL_CONFIG_DWORD_INFO(INTERNAL_EventPipeRundown, W("EventPipeRundown"), 1, "E
 RETAIL_CONFIG_DWORD_INFO(INTERNAL_EventPipeCircularMB, W("EventPipeCircularMB"), 1024, "The EventPipe circular buffer size in megabytes.")
 RETAIL_CONFIG_DWORD_INFO(INTERNAL_EventPipeProcNumbers, W("EventPipeProcNumbers"), 0, "Enable/disable capturing processor numbers in EventPipe event headers")
 RETAIL_CONFIG_DWORD_INFO(INTERNAL_EventPipeOutputStreaming, W("EventPipeOutputStreaming"), 1, "Enable/disable streaming for trace file set in DOTNET_EventPipeOutputPath.  Non-zero values enable streaming.")
+RETAIL_CONFIG_DWORD_INFO(INTERNAL_EventPipeBufferingMode, W("EventPipeBufferingMode"), 0, "Buffering mode for the DOTNET_EnableEventPipe startup session: 0 = Drop (default, lossy), 1 = Block (non-lossy); any other value falls back to Drop. Block applies only to streaming sessions (DOTNET_EventPipeOutputStreaming=1) and is ignored for non-streaming file sessions.")
 RETAIL_CONFIG_DWORD_INFO(INTERNAL_EventPipeEnableStackwalk, W("EventPipeEnableStackwalk"), 1, "Set to 0 to disable collecting stacks for EventPipe events.")
 RETAIL_CONFIG_DWORD_INFO(INTERNAL_EventPipeThreadSamplingRate, W("EventPipeThreadSamplingRate"), 0, "Desired sample interval in milliseconds for EventPipe thread time sampling profiler. 0 means use the default.")
 

src/coreclr/nativeaot/Runtime/eventpipe/ep-rt-aot.h

@@ -479,6 +479,23 @@ ep_rt_config_value_get_circular_mb (void)
     return 0;
 }
 
+static
+inline
+uint32_t
+ep_rt_config_value_get_buffering_mode (void)
+{
+    STATIC_CONTRACT_NOTHROW;
+
+    uint64_t value;
+    if (RhConfig::Environment::TryGetIntegerValue("EventPipeBufferingMode", &value))
+    {
+        EP_ASSERT(value <= UINT32_MAX);
+        return static_cast<uint32_t>(value);
+    }
+
+    return 0;
+}
+
 static
 inline
 bool
@@ -774,9 +791,19 @@ EP_RT_DEFINE_THREAD_FUNC (ep_rt_thread_aot_start_session_or_sampling_thread)
 
     ep_rt_thread_params_t* thread_params = reinterpret_cast<ep_rt_thread_params_t *>(data);
 
-    // We will create a new thread. cannot call ep_rt_aot_thread_get_handle since that will return null
-    extern ep_rt_thread_handle_t ep_rt_aot_setup_thread (void);
-    thread_params->thread = ep_rt_aot_setup_thread ();
+    if (thread_params->thread_type == EP_THREAD_TYPE_SESSION) {
+        // The session drain thread runs a purely native drain loop whose blocking primitives (PalSleep,
+        // CLREventStatic::Wait, CrstStatic::Enter) all tolerate a thread with no runtime Thread, so - like the
+        // CoreCLR native drain thread - it does not attach to the ThreadStore. That lets it start during
+        // diagnostic-port startup suspension, before RuntimeInstance/ThreadStore is initialized, without
+        // AttachCurrentThread dereferencing a not-yet-created RuntimeInstance.
+        thread_params->thread = NULL;
+    } else if (thread_params->thread_type == EP_THREAD_TYPE_SAMPLING) {
+        // The sampling thread's callback walks managed stacks, so it attaches to the ThreadStore via
+        // ep_rt_aot_setup_thread (ThreadStore::AttachCurrentThread).
+        extern ep_rt_thread_handle_t ep_rt_aot_setup_thread (void);
+        thread_params->thread = ep_rt_aot_setup_thread ();
+    }
 
     size_t result = thread_params->thread_func (thread_params);
     delete thread_params;

src/coreclr/nativeaot/Runtime/eventpipeinternal.cpp

@@ -96,7 +96,7 @@ EXTERN_C uint64_t QCALLTYPE EventPipeInternal_Enable(
         nullptr);
     ep_rt_utf8_string_free (outputPathUTF8);
 
-    ep_start_streaming(result);
+    ep_start_session(result);
 
     if (configProviders) {
         for (uint32_t i = 0; i < numProviders; ++i)

src/coreclr/vm/eventing/eventpipe/ep-rt-coreclr.h

@@ -564,6 +564,15 @@ ep_rt_config_value_get_circular_mb (void)
 	return CLRConfig::GetConfigValue (CLRConfig::INTERNAL_EventPipeCircularMB);
 }
 
+static
+inline
+uint32_t
+ep_rt_config_value_get_buffering_mode (void)
+{
+	STATIC_CONTRACT_NOTHROW;
+	return CLRConfig::GetConfigValue (CLRConfig::INTERNAL_EventPipeBufferingMode);
+}
+
 static
 inline
 bool
@@ -978,7 +987,39 @@ ep_rt_thread_create (
 				result = true;
 			}
 		}
-		else if (thread_type == EP_THREAD_TYPE_SESSION || thread_type == EP_THREAD_TYPE_SAMPLING)
+		else if (thread_type == EP_THREAD_TYPE_SESSION)
+		{
+			// Create the session drain thread as a raw native thread (no managed Thread), like the diagnostics
+			// server thread, so it never enters cooperative GC mode and can start during early startup before
+			// the GC / Thread Store are initialized - removing the need to defer session streaming until
+			// ep_finish_init. Unlike the SERVER branch it must carry the session pointer, so it wraps params
+			// and reuses ep_rt_thread_coreclr_start_func (which skips DestroyThread when thread == NULL).
+			rt_coreclr_thread_params_internal_t *thread_params = new (nothrow) rt_coreclr_thread_params_internal_t ();
+			if (thread_params)
+			{
+				thread_params->thread_params.thread_type = thread_type;
+				thread_params->thread_params.thread = NULL;
+				thread_params->thread_params.thread_func = reinterpret_cast<LPTHREAD_START_ROUTINE>(thread_func);
+				thread_params->thread_params.thread_params = params;
+
+				DWORD native_thread_id = 0;
+				HANDLE native_thread = ::CreateThread (nullptr, 0, reinterpret_cast<LPTHREAD_START_ROUTINE>(ep_rt_thread_coreclr_start_func), thread_params, 0, &native_thread_id);
+				if (native_thread != NULL)
+				{
+					if (id)
+					{
+						*reinterpret_cast<DWORD *>(id) = native_thread_id;
+					}
+					::CloseHandle (native_thread);
+					result = true;
+				}
+				else
+				{
+					delete thread_params;
+				}
+			}
+		}
+		else if (thread_type == EP_THREAD_TYPE_SAMPLING)
 		{
 			rt_coreclr_thread_params_internal_t *thread_params = new (nothrow) rt_coreclr_thread_params_internal_t ();
 			if (thread_params)

src/coreclr/vm/eventpipeadapter.h

@@ -254,7 +254,7 @@ class EventPipeAdapter final
 		}
 		CONTRACTL_END;
 
-		ep_start_streaming(id);
+		ep_start_session(id);
 	}
 
 	static inline EventPipeSession * GetSession(EventPipeSessionID id)

src/mono/mono/component/event_pipe.c

@@ -101,7 +101,7 @@ static MonoComponentEventPipe fn_table = {
 	&ep_disable,
 	&event_pipe_get_next_event,
 	&ep_get_wait_handle,
-	&ep_start_streaming,
+	&ep_start_session,
 	&ep_write_event_2,
 	&event_pipe_add_rundown_execution_checkpoint,
 	&event_pipe_add_rundown_execution_checkpoint_2,

src/mono/mono/eventpipe/ep-rt-mono.h

@@ -592,6 +592,15 @@ ep_rt_config_value_get_circular_mb (void)
 	return circular_mb;
 }
 
+static
+inline
+uint32_t
+ep_rt_config_value_get_buffering_mode (void)
+{
+	/* Unsupported */
+	return 0;
+}
+
 static
 inline
 bool
@@ -955,12 +964,33 @@ EP_RT_DEFINE_THREAD_FUNC (ep_rt_thread_mono_start_func)
 {
 	rt_mono_thread_params_internal_t *thread_params = (rt_mono_thread_params_internal_t *)data;
 
-	ep_rt_mono_thread_setup_2 (thread_params->background_thread, thread_params->thread_params.thread_type);
+	const EventPipeThreadType thread_type = thread_params->thread_params.thread_type;
+
+	if (thread_type == EP_THREAD_TYPE_SERVER) {
+		// The diagnostics server thread dispatches managed IPC command callbacks, so it takes a full managed attach.
+		ep_rt_mono_thread_setup_2 (thread_params->background_thread, thread_type);
+	} else if (thread_type == EP_THREAD_TYPE_SESSION) {
+		// The session drain thread runs only the native drain loop; it never runs managed code (managed provider
+		// callbacks auto-attach through the native->managed wrapper, and no managed provider can be registered
+		// before the runtime finishes starting up). Attach it at the thread-info level only - a MonoThreadInfo is
+		// all the drain loop's cooperative-GC primitives (sleep/wait/lock) need. A full managed attach's teardown
+		// runs mono_thread_internal_detach -> mono_gc_finalize_notify, which aborts when a session is started and
+		// stopped during diagnostic-port startup suspension, before the finalizer thread exists.
+		mono_thread_info_attach ();
+	} else if (thread_type == EP_THREAD_TYPE_SAMPLING) {
+		// The sample profiler thread walks managed stacks, so it takes a full managed attach.
+		ep_rt_mono_thread_setup_2 (thread_params->background_thread, thread_type);
+	}
 
 	thread_params->thread_params.thread = ep_rt_thread_get_handle ();
 	mono_thread_start_return_t result = thread_params->thread_params.thread_func (thread_params);
 
-	ep_rt_mono_thread_teardown ();
+	// Tear down symmetrically: only the SESSION thread's thread-info attach avoids the managed detach path
+	// (mono_thread_internal_detach -> mono_gc_finalize_notify).
+	if (thread_type == EP_THREAD_TYPE_SESSION)
+		mono_thread_info_detach ();
+	else
+		ep_rt_mono_thread_teardown ();
 
 	g_free (thread_params);
 

src/mono/mono/eventpipe/test/ep-buffer-manager-tests.c

@@ -10,6 +10,7 @@
 #include <eventpipe/ep-session.h>
 #include <eventpipe/ep-buffer-manager.h>
 #include <eventpipe/ep-file.h>
+#include <eventpipe/ep-thread.h>
 #include <eglib/test/test.h>
 
 #define TEST_PROVIDER_NAME "MyTestProvider"
@@ -66,8 +67,9 @@ buffer_manager_fini (
 
 static
 RESULT
-buffer_manager_init (
+buffer_manager_init_mode (
 	EventPipeSerializationFormat format,
+	EventPipeBufferingMode buffering_mode,
 	EventPipeBufferManager **buffer_manager,
 	ep_rt_thread_handle_t *thread_handle,
 	EventPipeThread **thread,
@@ -96,15 +98,17 @@ buffer_manager_init (
 			1,
 			TEST_FILE,
 			NULL,
-			EP_SESSION_TYPE_FILE,
+			(buffering_mode == EP_BUFFERING_MODE_BLOCK) ? EP_SESSION_TYPE_FILESTREAM : EP_SESSION_TYPE_FILE,
 			format,
+			0,
 			false,
 			1,
 			current_provider_config,
 			1,
 			NULL,
 			NULL,
-			0);
+			0,
+			buffering_mode);
 	EP_LOCK_EXIT (section1)
 
 	ep_raise_error_if_nok (*session != NULL);
@@ -149,6 +153,20 @@ buffer_manager_init (
 	ep_exit_error_handler ();
 }
 
+static
+RESULT
+buffer_manager_init (
+	EventPipeSerializationFormat format,
+	EventPipeBufferManager **buffer_manager,
+	ep_rt_thread_handle_t *thread_handle,
+	EventPipeThread **thread,
+	EventPipeSession **session,
+	EventPipeProvider **provider,
+	EventPipeEvent **ep_event)
+{
+	return buffer_manager_init_mode (format, EP_BUFFERING_MODE_DROP, buffer_manager, thread_handle, thread, session, provider, ep_event);
+}
+
 static
 bool
 write_events (
@@ -161,16 +179,26 @@ write_events (
 {
 	bool result = true;
 	uint32_t i = 0;
+
+	// ep_buffer_manager_write_event asserts the calling thread has published this session's index
+	// (the production write path sets it before writing); mirror that here so the assert holds.
+	EventPipeThread *current_thread = ep_thread_get_or_create ();
+	if (current_thread)
+		ep_thread_set_session_use_in_progress (current_thread, ep_session_get_index (session) | EP_SESSION_USE_WRITE_BUFFER_IN_USE);
+
 	for (; i < event_count; ++i) {
 		EventPipeEventPayload payload;
 		ep_event_payload_init (&payload, (uint8_t *)TEST_EVENT_DATA, ARRAY_SIZE (TEST_EVENT_DATA));
-		result = ep_buffer_manager_write_event (buffer_manager, thread, session, ep_event, &payload, NULL, NULL, thread, NULL);
+		result = ep_buffer_manager_write_event (buffer_manager, thread, session, ep_event, &payload, NULL, NULL, thread, NULL) == EP_WRITE_EVENT_RESULT_WRITTEN;
 		ep_event_payload_fini (&payload);
 
 		if (!result)
 			break;
 	}
 
+	if (current_thread)
+		ep_thread_set_session_use_in_progress (current_thread, UINT32_MAX);
+
 	if (events_written)
 		*events_written = i;
 	return result;
@@ -580,6 +608,89 @@ test_buffer_manager_perf (void)
 	ep_exit_error_handler ();
 }
 
+static RESULT
+test_buffer_manager_block_mode_abort_and_disable (void)
+{
+	RESULT result = NULL;
+	uint32_t test_location = 0;
+	EventPipeBufferManager *buffer_manager = NULL;
+	ep_rt_thread_handle_t thread_handle;
+	EventPipeThread *thread = NULL;
+	EventPipeSession *session = NULL;
+	EventPipeProvider *provider = NULL;
+	EventPipeEvent *ep_event = NULL;
+	EventPipeThread *current_thread = NULL;
+	bool use_in_progress_set = false;
+	bool observed_blocked = false;
+	uint32_t i = 0;
+
+	result = buffer_manager_init_mode (EP_SERIALIZATION_FORMAT_NETTRACE_V4, EP_BUFFERING_MODE_BLOCK, &buffer_manager, &thread_handle, &thread, &session, &provider, &ep_event);
+
+	ep_raise_error_if_nok (result == NULL);
+
+	test_location = 1;
+
+	ep_raise_error_if_nok (buffer_manager != NULL && session != NULL);
+
+	test_location = 2;
+
+	ep_raise_error_if_nok (ep_buffer_manager_get_buffering_mode (buffer_manager) == EP_BUFFERING_MODE_BLOCK);
+
+	test_location = 3;
+
+	// Publish this session's index, as a real producer does, so ep_buffer_manager_write_event's assert holds.
+	current_thread = ep_thread_get_or_create ();
+	ep_raise_error_if_nok (current_thread != NULL);
+	ep_thread_set_session_use_in_progress (current_thread, ep_session_get_index (session) | EP_SESSION_USE_WRITE_BUFFER_IN_USE);
+	use_in_progress_set = true;
+
+	test_location = 4;
+
+	// Fill the buffer pool. In Block mode a full buffer reports BLOCKED (park-and-retry) rather than
+	// dropping, so the producer never loses the event.
+	for (i = 0; i < 1000 * 1000; ++i) {
+		EventPipeEventPayload payload;
+		ep_event_payload_init (&payload, (uint8_t *)TEST_EVENT_DATA, ARRAY_SIZE (TEST_EVENT_DATA));
+		EventPipeWriteEventResult write_result = ep_buffer_manager_write_event (buffer_manager, thread_handle, session, ep_event, &payload, NULL, NULL, thread_handle, NULL);
+		ep_event_payload_fini (&payload);
+		if (write_result == EP_WRITE_EVENT_RESULT_BLOCKED) {
+			observed_blocked = true;
+			break;
+		}
+	}
+
+	ep_raise_error_if_nok (observed_blocked);
+
+	test_location = 5;
+
+	// Abort the blocked writers (the teardown step): the flag is raised so a parked producer gives up.
+	ep_buffer_manager_abort_blocked_writers (buffer_manager);
+	ep_raise_error_if_nok (ep_buffer_manager_is_aborting (buffer_manager));
+
+	test_location = 6;
+
+	// With abort raised, a write on the still-full buffer now drops (gives up) instead of parking.
+	{
+		EventPipeEventPayload payload;
+		ep_event_payload_init (&payload, (uint8_t *)TEST_EVENT_DATA, ARRAY_SIZE (TEST_EVENT_DATA));
+		EventPipeWriteEventResult write_result = ep_buffer_manager_write_event (buffer_manager, thread_handle, session, ep_event, &payload, NULL, NULL, thread_handle, NULL);
+		ep_event_payload_fini (&payload);
+		ep_raise_error_if_nok (write_result != EP_WRITE_EVENT_RESULT_BLOCKED);
+	}
+
+ep_on_exit:
+	if (use_in_progress_set && current_thread != NULL)
+		ep_thread_set_session_use_in_progress (current_thread, UINT32_MAX);
+	// buffer_manager_fini -> ep_session_dec_ref exercises tearing down a Block-mode session.
+	buffer_manager_fini (buffer_manager, thread, session, provider, ep_event);
+	return result;
+
+ep_on_error:
+	if (!result)
+		result = FAILED ("Failed at test location=%i", test_location);
+	ep_exit_error_handler ();
+}
+
 static RESULT
 test_buffer_manager_teardown (void)
 {
@@ -608,6 +719,7 @@ static Test ep_buffer_manager_tests [] = {
 	{"test_buffer_manager_write_events_to_file_v3", test_buffer_manager_write_events_to_file_v3},
 	{"test_buffer_manager_write_events_to_file_v4", test_buffer_manager_write_events_to_file_v4},
 	{"test_buffer_manager_oom", test_buffer_manager_oom},
+	{"test_buffer_manager_block_mode_abort_and_disable", test_buffer_manager_block_mode_abort_and_disable},
 #ifdef TEST_PERF
 	{"test_buffer_manager_perf", test_buffer_manager_perf},
 #endif