[EventPipe] End user_events session upon continuation_stream closure

src/native/eventpipe/ds-eventpipe-protocol.c

@@ -920,7 +920,7 @@ eventpipe_protocol_helper_collect_tracing (
 		payload->serialization_format,
 		payload->rundown_keyword,
 		payload->stackwalk_requested,
-		payload->session_type == EP_SESSION_TYPE_IPCSTREAM ? ds_ipc_stream_get_stream_ref (stream) : NULL,
+		ds_ipc_stream_get_stream_ref (stream),
 		NULL,
 		NULL,
 		user_events_data_fd);

src/native/eventpipe/ds-ipc-pal-namedpipe.c

@@ -213,7 +213,7 @@ ds_ipc_poll (
 			handles [i] = poll_handles_data [i].ipc->overlap.hEvent;
 			if (handles [i] == INVALID_HANDLE_VALUE) {
 				// Invalid handle, wait will fail. Signal error
-				poll_handles_data [i].events = DS_IPC_POLL_EVENTS_ERR;
+				poll_handles_data [i].events = EP_IPC_POLL_EVENTS_ERR;
 			}
 		} else {
 			// CLIENT
@@ -238,7 +238,7 @@ ds_ipc_poll (
 						handles [i] = poll_handles_data [i].stream->overlap.hEvent;
 						break;
 					case ERROR_PIPE_NOT_CONNECTED:
-						poll_handles_data [i].events = (uint8_t)DS_IPC_POLL_EVENTS_HANGUP;
+						poll_handles_data [i].events = (uint8_t)EP_IPC_POLL_EVENTS_HANGUP;
 						result = -1;
 						ep_raise_error ();
 					default:
@@ -288,7 +288,7 @@ ds_ipc_poll (
 		// check if we abandoned something
 		DWORD abandonedIndex = wait - WAIT_ABANDONED_0;
 		if (abandonedIndex > 0 || abandonedIndex < (poll_handles_data_len - 1)) {
-			poll_handles_data [abandonedIndex].events = (uint8_t)DS_IPC_POLL_EVENTS_HANGUP;
+			poll_handles_data [abandonedIndex].events = (uint8_t)EP_IPC_POLL_EVENTS_HANGUP;
 			result = -1;
 			ep_raise_error ();
 		} else {
@@ -325,20 +325,20 @@ ds_ipc_poll (
 		if (!success) {
 			DWORD error = GetLastError();
 			if (error == ERROR_PIPE_NOT_CONNECTED || error == ERROR_BROKEN_PIPE) {
-				poll_handles_data [index].events = (uint8_t)DS_IPC_POLL_EVENTS_HANGUP;
+				poll_handles_data [index].events = (uint8_t)EP_IPC_POLL_EVENTS_HANGUP;
 			} else {
 				if (callback)
 					callback ("Client connection error", error);
-				poll_handles_data [index].events = (uint8_t)DS_IPC_POLL_EVENTS_ERR;
+				poll_handles_data [index].events = (uint8_t)EP_IPC_POLL_EVENTS_ERR;
 				result = -1;
 				ep_raise_error ();
 			}
 		} else {
-			poll_handles_data [index].events = (uint8_t)DS_IPC_POLL_EVENTS_SIGNALED;
+			poll_handles_data [index].events = (uint8_t)EP_IPC_POLL_EVENTS_SIGNALED;
 		}
 	} else {
 		// SERVER
-		poll_handles_data [index].events = (uint8_t)DS_IPC_POLL_EVENTS_SIGNALED;
+		poll_handles_data [index].events = (uint8_t)EP_IPC_POLL_EVENTS_SIGNALED;
 	}
 
 	result = 1;
@@ -834,12 +834,23 @@ ipc_stream_close_func (void *object)
 	return ds_ipc_stream_close (ipc_stream, NULL);
 }
 
+static
+EventPipeIpcPollEvents
+ipc_stream_poll_func (
+	void *object,
+	uint32_t timeout_ms)
+{
+	// Needs to be implemented.
+	return EP_IPC_POLL_EVENTS_UNKNOWN;
+}
+
 static IpcStreamVtable ipc_stream_vtable = {
 	ipc_stream_free_func,
 	ipc_stream_read_func,
 	ipc_stream_write_func,
 	ipc_stream_flush_func,
-	ipc_stream_close_func };
+	ipc_stream_close_func,
+	ipc_stream_poll_func };
 
 static
 DiagnosticsIpcStream *

src/native/eventpipe/ds-ipc-pal-socket.c

@@ -1146,15 +1146,15 @@ ds_ipc_poll (
 				// check for hangup first because a closed socket
 				// will technically meet the requirements for POLLIN
 				// i.e., a call to recv/read won't block
-				poll_handles_data [i].events = (uint8_t)DS_IPC_POLL_EVENTS_HANGUP;
+				poll_handles_data [i].events = (uint8_t)EP_IPC_POLL_EVENTS_HANGUP;
 			} else if ((poll_fds [i].revents & (POLLERR|POLLNVAL))) {
 				if (callback)
 					callback ("Poll error", (uint32_t)poll_fds [i].revents);
-				poll_handles_data [i].events = (uint8_t)DS_IPC_POLL_EVENTS_ERR;
+				poll_handles_data [i].events = (uint8_t)EP_IPC_POLL_EVENTS_ERR;
 			} else if (poll_fds [i].revents & (POLLIN|POLLPRI)) {
-				poll_handles_data [i].events = (uint8_t)DS_IPC_POLL_EVENTS_SIGNALED;
+				poll_handles_data [i].events = (uint8_t)EP_IPC_POLL_EVENTS_SIGNALED;
 			} else {
-				poll_handles_data [i].events = (uint8_t)DS_IPC_POLL_EVENTS_UNKNOWN;
+				poll_handles_data [i].events = (uint8_t)EP_IPC_POLL_EVENTS_UNKNOWN;
 				if (callback)
 					callback ("unknown poll response", (uint32_t)poll_fds [i].revents);
 			}
@@ -1489,12 +1489,24 @@ ipc_stream_close_func (void *object)
 	return ds_ipc_stream_close (ipc_stream, NULL);
 }
 
+static
+EventPipeIpcPollEvents
+ipc_stream_poll_func (
+	void *object,
+	uint32_t timeout_ms)
+{
+	EP_ASSERT (object != NULL);
+	DiagnosticsIpcStream *ipc_stream = (DiagnosticsIpcStream *)object;
+	return ds_ipc_stream_poll (ipc_stream, timeout_ms);
+}
+
 static IpcStreamVtable ipc_stream_vtable = {
 	ipc_stream_free_func,
 	ipc_stream_read_func,
 	ipc_stream_write_func,
 	ipc_stream_flush_func,
-	ipc_stream_close_func };
+	ipc_stream_close_func,
+	ipc_stream_poll_func };
 
 static
 DiagnosticsIpcStream *
@@ -1668,6 +1680,44 @@ ds_ipc_stream_to_string (
 	return (result > 0 && result < (int32_t)buffer_len) ? result : 0;
 }
 
+EventPipeIpcPollEvents
+ds_ipc_stream_poll (
+	DiagnosticsIpcStream *ipc_stream,
+	uint32_t timeout_ms)
+{
+	EP_ASSERT (ipc_stream != NULL);
+
+	if (ipc_stream->client_socket == DS_IPC_INVALID_SOCKET)
+		return EP_IPC_POLL_EVENTS_HANGUP;
+
+	ds_ipc_pollfd_t pfd;
+	pfd.fd = ipc_stream->client_socket;
+	pfd.events = POLLIN | POLLPRI | POLLOUT;
+
+	int result_poll;
+	result_poll = ipc_poll_fds (&pfd, 1, timeout_ms);
+
+	if (result_poll < 0)
+		return EP_IPC_POLL_EVENTS_ERR;
+
+	if (result_poll == 0)
+		return EP_IPC_POLL_EVENTS_HANGUP;
+
+	if (pfd.revents == 0)
+		return EP_IPC_POLL_EVENTS_NONE;
+
+	if (pfd.revents & POLLHUP)
+		return EP_IPC_POLL_EVENTS_HANGUP;
+
+	if (pfd.revents & (POLLERR | POLLNVAL))
+		return EP_IPC_POLL_EVENTS_ERR;
+
+	if (pfd.revents & (POLLIN | POLLPRI | POLLOUT))
+		return EP_IPC_POLL_EVENTS_SIGNALED;
+
+	return EP_IPC_POLL_EVENTS_UNKNOWN;
+}
+
 #endif /* ENABLE_PERFTRACING */
 
 #ifndef DS_INCLUDE_SOURCE_FILES

src/native/eventpipe/ds-ipc-pal-types.h

@@ -22,14 +22,6 @@ typedef struct _DiagnosticsIpcStream DiagnosticsIpcStream;
  * Diagnostics IPC PAL Enums.
  */
 
-typedef enum {
-	DS_IPC_POLL_EVENTS_NONE = 0x00, // no events
-	DS_IPC_POLL_EVENTS_SIGNALED = 0x01, // ready for use
-	DS_IPC_POLL_EVENTS_HANGUP = 0x02, // connection remotely closed
-	DS_IPC_POLL_EVENTS_ERR = 0x04, // error
-	DS_IPC_POLL_EVENTS_UNKNOWN = 0x80 // unknown state
-} DiagnosticsIpcPollEvents;
-
 typedef enum {
 	DS_IPC_CONNECTION_MODE_CONNECT,
 	DS_IPC_CONNECTION_MODE_LISTEN

src/native/eventpipe/ds-ipc-pal-websocket.c

@@ -248,11 +248,11 @@ ds_ipc_poll (
 		int client_socket = poll_handles_data [i].stream->client_socket;
 		int pending = ds_rt_websocket_poll (client_socket);
 		if (pending < 0){
-			poll_handles_data [i].events = (uint8_t)DS_IPC_POLL_EVENTS_ERR;
+			poll_handles_data [i].events = (uint8_t)EP_IPC_POLL_EVENTS_ERR;
 			return 1;
 		}
 		if (pending > 0){
-			poll_handles_data [i].events = (uint8_t)DS_IPC_POLL_EVENTS_SIGNALED;
+			poll_handles_data [i].events = (uint8_t)EP_IPC_POLL_EVENTS_SIGNALED;
 			return 1;
 		}
 	}
@@ -425,12 +425,33 @@ ipc_stream_close_func (void *object)
 	return ds_ipc_stream_close (ipc_stream, NULL);
 }
 
+static
+EventPipeIpcPollEvents
+ipc_stream_poll_func (
+	void *object,
+	uint32_t timeout_ms)
+{
+	EP_ASSERT (object != NULL);
+	DiagnosticsIpcStream *ipc_stream = (DiagnosticsIpcStream *)object;
+
+	// Check if the socket is still open
+	int pending = ds_rt_websocket_poll (ipc_stream->client_socket);
+	if (pending < 0)
+		return EP_IPC_POLL_EVENTS_ERR;
+
+	if (pending > 0)
+		return EP_IPC_POLL_EVENTS_SIGNALED;
+
+	return EP_IPC_POLL_EVENTS_NONE;
+}
+
 static IpcStreamVtable ipc_stream_vtable = {
 	ipc_stream_free_func,
 	ipc_stream_read_func,
 	ipc_stream_write_func,
 	ipc_stream_flush_func,
-	ipc_stream_close_func };
+	ipc_stream_close_func,
+	ipc_stream_poll_func };
 
 static
 DiagnosticsIpcStream *

src/native/eventpipe/ds-ipc-pal.h

@@ -140,5 +140,10 @@ ds_ipc_stream_to_string (
 	ep_char8_t *buffer,
 	uint32_t buffer_len);
 
+EventPipeIpcPollEvents
+ds_ipc_stream_poll (
+	DiagnosticsIpcStream *ipc_stream,
+	uint32_t timeout_ms);
+
 #endif /* ENABLE_PERFTRACING */
 #endif /* __DIAGNOSTICS_IPC_PAL_H__ */

src/native/eventpipe/ds-ipc.c

@@ -406,13 +406,13 @@ ds_ipc_stream_factory_get_next_available_stream (ds_ipc_error_callback_func call
 			DN_VECTOR_FOREACH_BEGIN (DiagnosticsIpcPollHandle, ipc_poll_handle, &ipc_poll_handles) {
 				DiagnosticsPort *port = (DiagnosticsPort *)ipc_poll_handle.user_data;
 				switch (ipc_poll_handle.events) {
-				case DS_IPC_POLL_EVENTS_HANGUP:
+				case EP_IPC_POLL_EVENTS_HANGUP:
 					EP_ASSERT (port != NULL);
 					ds_port_reset_vcall (port, callback);
 					DS_LOG_INFO_2 ("ds_ipc_stream_factory_get_next_available_stream - HUP :: Poll attempt: %d, connection %d hung up. Connect is reset.", poll_attempts, connection_id);
 					poll_timeout_ms = DS_IPC_POLL_TIMEOUT_MIN_MS;
 					break;
-				case DS_IPC_POLL_EVENTS_SIGNALED:
+				case EP_IPC_POLL_EVENTS_SIGNALED:
 					EP_ASSERT (port != NULL);
 					if (!stream) {  // only use first signaled stream; will get others on subsequent calls
 						stream = ds_port_get_connected_stream_vcall (port, callback);
@@ -422,12 +422,12 @@ ds_ipc_stream_factory_get_next_available_stream (ds_ipc_error_callback_func call
 					}
 					DS_LOG_DEBUG_2 ("ds_ipc_stream_factory_get_next_available_stream - SIG :: Poll attempt: %d, connection %d signalled.", poll_attempts, connection_id);
 					break;
-				case DS_IPC_POLL_EVENTS_ERR:
+				case EP_IPC_POLL_EVENTS_ERR:
 					ds_port_reset_vcall ((DiagnosticsPort *)ipc_poll_handle.user_data, callback);
 					DS_LOG_INFO_2 ("ds_ipc_stream_factory_get_next_available_stream - ERR :: Poll attempt: %d, connection %d errored. Connection is reset.", poll_attempts, connection_id);
 					saw_error = true;
 					break;
-				case DS_IPC_POLL_EVENTS_NONE:
+				case EP_IPC_POLL_EVENTS_NONE:
 					DS_LOG_INFO_2 ("ds_ipc_stream_factory_get_next_available_stream - NON :: Poll attempt: %d, connection %d had no events.", poll_attempts, connection_id);
 					break;
 				default:

src/native/eventpipe/ep-ipc-pal-types.h

@@ -11,5 +11,19 @@
 
 #include "ep-ipc-pal-types-forward.h"
 
+/*
+ *  EventPipe IPC PAL
+ */
+
+typedef enum {
+	EP_IPC_POLL_EVENTS_NONE = 0x00, // no events
+	EP_IPC_POLL_EVENTS_SIGNALED = 0x01, // ready for use
+	EP_IPC_POLL_EVENTS_HANGUP = 0x02, // connection remotely closed
+	EP_IPC_POLL_EVENTS_ERR = 0x04, // error
+	EP_IPC_POLL_EVENTS_UNKNOWN = 0x80 // unknown state
+} EventPipeIpcPollEvents;
+
+#define EP_IPC_POLL_TIMEOUT_MIN_MS (uint32_t)10
+
 #endif /* ENABLE_PERFTRACING */
 #endif /* __EVENTPIPE_IPC_PAL_TYPES_H__ */

src/native/eventpipe/ep-ipc-stream.h

@@ -21,13 +21,15 @@ typedef bool (*IpcStreamReadFunc)(void *object, uint8_t *buffer, uint32_t bytes_
 typedef bool (*IpcStreamWriteFunc)(void *object, const uint8_t *buffer, uint32_t bytes_to_write, uint32_t *bytes_written, uint32_t timeout_ms);
 typedef bool (*IpcStreamFlushFunc)(void *object);
 typedef bool (*IpcStreamCloseFunc)(void *object);
+typedef EventPipeIpcPollEvents (*IpcStreamPollFunc)(void *object, uint32_t timeout_ms);
 
 struct _IpcStreamVtable {
 	IpcStreamFreeFunc free_func;
 	IpcStreamReadFunc read_func;
 	IpcStreamWriteFunc write_func;
 	IpcStreamFlushFunc flush_func;
 	IpcStreamCloseFunc close_func;
+	IpcStreamPollFunc poll_func;
 };
 
 #if defined(EP_INLINE_GETTER_SETTER) || defined(EP_IMPL_IPC_STREAM_GETTER_SETTER) || defined(DS_IMPL_IPC_PAL_NAMEDPIPE_GETTER_SETTER) || defined(DS_IMPL_IPC_PAL_SOCKET_GETTER_SETTER)
@@ -77,5 +79,8 @@ ep_ipc_stream_flush_vcall (IpcStream *ipc_stream);
 bool
 ep_ipc_stream_close_vcall (IpcStream *ipc_stream);
 
+EventPipeIpcPollEvents
+ep_ipc_stream_poll_vcall (IpcStream *ipc_stream, uint32_t timeout_ms);
+
 #endif /* ENABLE_PERFTRACING */
 #endif /* __EVENTPIPE_IPC_STREAM_H__ */