[wasm-ep] Implement DiagnosticServer and startup sessions for WebAssembly

src/mono/mono.proj

@@ -483,6 +483,7 @@
     <ItemGroup Condition="'$(TargetsBrowser)' == 'true'">
       <_MonoCMakeArgs Include="-DFEATURE_PERFTRACING_DISABLE_PERFTRACING_LISTEN_PORTS=1"/>
       <_MonoCMakeArgs Include="-DFEATURE_PERFTRACING_DISABLE_DEFAULT_LISTEN_PORT=1"/>
+      <_MonoCMakeArgs Include="-DFEATURE_PERFTRACING_DISABLE_CONNECT_PORTS=1" />
     </ItemGroup>
 
     <!-- Components -->

src/mono/mono/component/diagnostics_server.c

@@ -7,10 +7,19 @@
 #include <mono/utils/mono-publib.h>
 #include <mono/utils/mono-compiler.h>
 #include <eventpipe/ds-server.h>
+#ifdef HOST_WASM
+#include <eventpipe/ep-ipc-stream.h>
+#include <mono/component/event_pipe-wasm.h>
+#include <mono/utils/mono-coop-semaphore.h>
+#include <mono/utils/mono-threads-wasm.h>
+#include <emscripten/emscripten.h>
+#include <emscripten/threading.h>
+#endif
 
 static bool
 diagnostics_server_available (void);
 
+#if !defined (HOST_WASM) || defined (DISABLE_THREADS)
 static MonoComponentDiagnosticsServer fn_table = {
 	{ MONO_COMPONENT_ITF_VERSION, &diagnostics_server_available },
 	&ds_server_init,
@@ -19,6 +28,291 @@ static MonoComponentDiagnosticsServer fn_table = {
 	&ds_server_disable
 };
 
+#else /* !defined (HOST_WASM) || defined (DISABLE_THREADS) */
+
+static bool
+ds_server_wasm_init (void);
+
+static bool
+ds_server_wasm_shutdown (void);
+
+static void
+ds_server_wasm_pause_for_diagnostics_monitor (void);
+
+static void
+ds_server_wasm_disable (void);
+
+static MonoComponentDiagnosticsServer fn_table = {
+	{ MONO_COMPONENT_ITF_VERSION, &diagnostics_server_available },
+	&ds_server_wasm_init,
+	&ds_server_wasm_shutdown,
+	&ds_server_wasm_pause_for_diagnostics_monitor,
+	&ds_server_wasm_disable,
+};
+
+typedef struct _MonoWasmDiagnosticServerOptions {
+	int32_t suspend; /* set from JS! */
+	MonoCoopSem suspend_resume;
+} MonoWasmDiagnosticServerOptions;
+
+static MonoWasmDiagnosticServerOptions wasm_ds_options;
+static pthread_t ds_thread_id;
+
+extern void
+mono_wasm_diagnostic_server_on_runtime_server_init (MonoWasmDiagnosticServerOptions *out_options);
+
+EMSCRIPTEN_KEEPALIVE void
+mono_wasm_diagnostic_server_resume_runtime_startup (void);
+
+static bool
+ds_server_wasm_init (void)
+{
+	/* called on the main thread when the runtime is sufficiently initialized */
+	mono_coop_sem_init (&wasm_ds_options.suspend_resume, 0);
+	mono_wasm_diagnostic_server_on_runtime_server_init(&wasm_ds_options);
+	return true;
+}
+
+
+static bool
+ds_server_wasm_shutdown (void)
+{
+	return true;
+}
+
+static void
+ds_server_wasm_pause_for_diagnostics_monitor (void)
+{
+	/* wait until the DS receives a resume */
+	if (wasm_ds_options.suspend) {
+		const guint timeout = 50;
+		const guint warn_threshold = 5000;
+		guint cumulative_timeout = 0;
+		while (true) {
+			MonoSemTimedwaitRet res = mono_coop_sem_timedwait (&wasm_ds_options.suspend_resume, timeout, MONO_SEM_FLAGS_ALERTABLE);
+			if (res == MONO_SEM_TIMEDWAIT_RET_SUCCESS || res == MONO_SEM_TIMEDWAIT_RET_ALERTED)
+				break;
+			else {
+				/* timed out */
+				cumulative_timeout += timeout;
+				if (cumulative_timeout > warn_threshold) {
+					cumulative_timeout = 0;
+				}
+			}
+		}
+	}
+}
+
+
+static void
+ds_server_wasm_disable (void)
+{
+}
+
+/* Allocated by mono_wasm_diagnostic_server_create_thread,
+ * then ownership passed to server_thread.
+ */
+static char*
+ds_websocket_url;
+
+extern void mono_wasm_diagnostic_server_on_server_thread_created (char *websocket_url);
+
+static void*
+server_thread (void* unused_arg G_GNUC_UNUSED)
+{
+	char* ws_url = g_strdup (ds_websocket_url);
+	g_free (ds_websocket_url);
+	ds_websocket_url = NULL;
+	mono_wasm_diagnostic_server_on_server_thread_created (ws_url);
+	// "exit" from server_thread, but keep the pthread alive and responding to events
+	emscripten_exit_with_live_runtime ();
+}
+
+gboolean
+mono_wasm_diagnostic_server_create_thread (const char *websocket_url, pthread_t *out_thread_id)
+{
+	pthread_t thread;
+
+	g_assert (!ds_websocket_url);
+	ds_websocket_url = g_strdup (websocket_url);
+	if (!pthread_create (&thread, NULL, server_thread, NULL)) {
+		*out_thread_id = thread;
+		return TRUE;
+	}
+	memset(out_thread_id, 0, sizeof(pthread_t));
+	return FALSE;
+}
+
+void
+mono_wasm_diagnostic_server_thread_attach_to_runtime (void)
+{
+	ds_thread_id = pthread_self();
+	MonoThread *thread = mono_thread_internal_attach (mono_get_root_domain ());
+	mono_thread_set_state (thread, ThreadState_Background);
+	mono_thread_info_set_flags (MONO_THREAD_INFO_FLAGS_NO_SAMPLE);
+	/* diagnostic server thread is now in GC Unsafe mode */
+}
+
+void
+mono_wasm_diagnostic_server_post_resume_runtime (void)
+{
+	if (wasm_ds_options.suspend) {
+		/* wake the main thread */
+		mono_coop_sem_post (&wasm_ds_options.suspend_resume);
+	}
+}
+
+#define QUEUE_CLOSE_SENTINEL ((uint8_t*)(intptr_t)-1)
+
+/* single-reader single-writer one-element queue. See
+ * src/mono/wasm/runtime/diagnostics/server_pthread/stream-queue.ts
+ */
+typedef struct WasmIpcStreamQueue {
+	uint8_t *buf; /* or QUEUE_CLOSE_SENTINEL */
+	int32_t count;
+	volatile int32_t buf_full;
+} WasmIpcStreamQueue;
+
+extern void
+mono_wasm_diagnostic_server_stream_signal_work_available (WasmIpcStreamQueue *queue, int32_t current_thread);
+
+static void
+queue_wake_reader (void *ptr) {
+	/* asynchronously invoked on the ds server thread by the writer. */
+	WasmIpcStreamQueue *q = (WasmIpcStreamQueue *)ptr;
+	mono_wasm_diagnostic_server_stream_signal_work_available (q, 0);
+}
+
+static void
+queue_wake_reader_now (WasmIpcStreamQueue *q)
+{
+	// call only from the diagnostic server thread!
+	mono_wasm_diagnostic_server_stream_signal_work_available (q, 1);
+}
+
+static int32_t
+queue_push_sync (WasmIpcStreamQueue *q, const uint8_t *buf, uint32_t buf_size, uint32_t *bytes_written)
+{
+	/* to be called on the writing thread */
+	/* single-writer, so there is no write contention */
+	q->buf = (uint8_t*)buf;
+	q->count = buf_size;
+	/* there's one instance where a thread other than the
+	 * streaming thread is writing: in ep_file_initialize_file
+	 * (called from ep_session_start_streaming), there's a write
+	 * from either the main thread (if the streaming was deferred
+	 * until ep_finish_init is called) or the diagnostic thread if
+	 * the session is started later.
+	 */
+	pthread_t cur = pthread_self ();
+	gboolean will_wait = TRUE;
+	mono_atomic_store_i32 (&q->buf_full, 1);
+	if (cur == ds_thread_id) {
+		queue_wake_reader_now (q);
+		/* doesn't return until the buffer is empty again; no need to wait */
+		will_wait = FALSE;
+	} else {
+		emscripten_dispatch_to_thread (ds_thread_id, EM_FUNC_SIG_VI, &queue_wake_reader, NULL, q);
+	}
+	// wait until the reader reads the value
+	int r = 0;
+	if (G_LIKELY (will_wait)) {
+		gboolean is_browser_thread_inited = FALSE;
+		gboolean is_browser_thread = FALSE;
+		while (mono_atomic_load_i32 (&q->buf_full) != 0) {
+			if (G_UNLIKELY (is_browser_thread_inited)) {
+					is_browser_thread = mono_threads_wasm_is_browser_thread ();
+					is_browser_thread_inited = TRUE;
+			}
+			if (G_UNLIKELY (is_browser_thread)) {
+				/* can't use memory.atomic.wait32 on the main thread, spin instead */
+				/* this lets Emscripten run queued calls on the main thread */
+				emscripten_thread_sleep (1);
+			} else  {
+				r = mono_wasm_atomic_wait_i32 (&q->buf_full, 1, -1);
+				if (G_UNLIKELY (r == 2)) {
+					/* timed out with infinite wait?? */
+					return -1;
+				}
+				/* if r == 0 (blocked and woken) or r == 1 (not equal), go around again and check if buf_full is now 0 */
+			}
+		}
+	}
+	if (bytes_written)
+		*bytes_written = buf_size;
+	return 0;
+}
+
+typedef struct {
+	IpcStream stream;
+	WasmIpcStreamQueue queue;
+}  WasmIpcStream;
+
+static void
+wasm_ipc_stream_free (void *self);
+static bool
+wasm_ipc_stream_read (void *self, uint8_t *buffer, uint32_t bytes_to_read, uint32_t *bytes_read, uint32_t timeout_ms);
+static bool
+wasm_ipc_stream_write (void *self, const uint8_t *buffer, uint32_t bytes_to_write, uint32_t *bytes_written, uint32_t timeout_ms);
+static bool
+wasm_ipc_stream_flush (void *self);
+static bool
+wasm_ipc_stream_close (void *self);
+
+static IpcStreamVtable wasm_ipc_stream_vtable = {
+	&wasm_ipc_stream_free,
+	&wasm_ipc_stream_read,
+	&wasm_ipc_stream_write,
+	&wasm_ipc_stream_flush,
+	&wasm_ipc_stream_close,
+};
+
+EMSCRIPTEN_KEEPALIVE IpcStream *
+mono_wasm_diagnostic_server_create_stream (void)
+{
+	g_assert (G_STRUCT_OFFSET(WasmIpcStream, queue) == 4); // keep in sync with mono_wasm_diagnostic_server_get_stream_queue
+	WasmIpcStream *stream = g_new0 (WasmIpcStream, 1);
+	ep_ipc_stream_init (&stream->stream, &wasm_ipc_stream_vtable);
+	return &stream->stream;
+}
+
+static void
+wasm_ipc_stream_free (void *self)
+{
+	g_free (self);
+}
+static bool
+wasm_ipc_stream_read (void *self, uint8_t *buffer, uint32_t bytes_to_read, uint32_t *bytes_read, uint32_t timeout_ms)
+{
+	/* our reader is in JS */
+	g_assert_not_reached();
+}
+static bool
+wasm_ipc_stream_write (void *self, const uint8_t *buffer, uint32_t bytes_to_write, uint32_t *bytes_written, uint32_t timeout_ms)
+{
+	WasmIpcStream *stream = (WasmIpcStream *)self;
+	g_assert (timeout_ms == EP_INFINITE_WAIT); // pass it down to the queue if the timeout param starts being used
+	int r = queue_push_sync (&stream->queue, buffer, bytes_to_write, bytes_written);
+	return r == 0;
+}
+
+static bool
+wasm_ipc_stream_flush (void *self)
+{
+	return true;
+}
+
+static bool
+wasm_ipc_stream_close (void *self)
+{
+	WasmIpcStream *stream = (WasmIpcStream*)self;
+	// push the special buf value -1 to signal stream close.
+	int r = queue_push_sync (&stream->queue, QUEUE_CLOSE_SENTINEL, 0, NULL);
+	return r == 0;
+}
+
+#endif /* !defined (HOST_WASM) || defined (DISABLE_THREADS) */
+
 static bool
 diagnostics_server_available (void)
 {
@@ -29,4 +323,5 @@ MonoComponentDiagnosticsServer *
 mono_component_diagnostics_server_init (void)
 {
 	return &fn_table;
+
 }

src/mono/mono/component/event_pipe-stub.c

@@ -6,9 +6,6 @@
 #include "mono/component/event_pipe.h"
 #include "mono/component/event_pipe-wasm.h"
 #include "mono/metadata/components.h"
-#ifdef HOST_WASM
-#include <emscripten/emscripten.h>
-#endif
 
 static EventPipeSessionID _dummy_session_id;
 
@@ -524,12 +521,12 @@ mono_component_event_pipe_init (void)
 
 EMSCRIPTEN_KEEPALIVE gboolean
 mono_wasm_event_pipe_enable (const ep_char8_t *output_path,
+			     IpcStream *ipc_stream,
 			     uint32_t circular_buffer_size_in_mb,
 			     const ep_char8_t *providers,
 			     /* EventPipeSessionType session_type = EP_SESSION_TYPE_FILE, */
 			     /* EventPipieSerializationFormat format = EP_SERIALIZATION_FORMAT_NETTRACE_V4, */
 			     /* bool */ gboolean rundown_requested,
-			     /* IpcStream stream = NULL, */
 			     /* EventPipeSessionSycnhronousCallback sync_callback = NULL, */
 			     /* void *callback_additional_data, */
 			     MonoWasmEventPipeSessionID *out_session_id)
@@ -551,5 +548,4 @@ mono_wasm_event_pipe_session_disable (MonoWasmEventPipeSessionID session_id)
 {
 	g_assert_not_reached ();
 }
-
 #endif /* HOST_WASM */

src/mono/mono/component/event_pipe-wasm.h

@@ -12,6 +12,7 @@
 
 #ifdef HOST_WASM
 
+#include <pthread.h>
 #include <emscripten.h>
 
 G_BEGIN_DECLS
@@ -27,12 +28,12 @@ typedef uint32_t MonoWasmEventPipeSessionID;
 
 EMSCRIPTEN_KEEPALIVE gboolean
 mono_wasm_event_pipe_enable (const ep_char8_t *output_path,
+			     IpcStream *ipc_stream,
 			     uint32_t circular_buffer_size_in_mb,
 			     const ep_char8_t *providers,
 			     /* EventPipeSessionType session_type = EP_SESSION_TYPE_FILE, */
 			     /* EventPipieSerializationFormat format = EP_SERIALIZATION_FORMAT_NETTRACE_V4, */
 			     /* bool */ gboolean rundown_requested,
-			     /* IpcStream stream = NULL, */
 			     /* EventPipeSessionSycnhronousCallback sync_callback = NULL, */
 			     /* void *callback_additional_data, */
 			     MonoWasmEventPipeSessionID *out_session_id);
@@ -43,6 +44,18 @@ mono_wasm_event_pipe_session_start_streaming (MonoWasmEventPipeSessionID session
 EMSCRIPTEN_KEEPALIVE gboolean
 mono_wasm_event_pipe_session_disable (MonoWasmEventPipeSessionID session_id);
 
+EMSCRIPTEN_KEEPALIVE gboolean
+mono_wasm_diagnostic_server_create_thread (const char *websocket_url, pthread_t *out_thread_id);
+
+EMSCRIPTEN_KEEPALIVE void
+mono_wasm_diagnostic_server_thread_attach_to_runtime (void);
+
+EMSCRIPTEN_KEEPALIVE void
+mono_wasm_diagnostic_server_post_resume_runtime (void);
+
+EMSCRIPTEN_KEEPALIVE IpcStream *
+mono_wasm_diagnostic_server_create_stream (void);
+
 G_END_DECLS
 
 #endif /* HOST_WASM */