Address Feedback

Revert moving IpcPollEvents to ep-ipc-pal-types header
Use IpcStream base type directly in sessions
Omit websocket ipc_stream_poll_func implementation
Change continuation stream poll timeout to infinite
Change returned PollEvent to none upon timeout

Author: mdh1418

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 = EP_IPC_POLL_EVENTS_ERR;
+				poll_handles_data [i].events = DS_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)EP_IPC_POLL_EVENTS_HANGUP;
+						poll_handles_data [i].events = (uint8_t)DS_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)EP_IPC_POLL_EVENTS_HANGUP;
+			poll_handles_data [abandonedIndex].events = (uint8_t)DS_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)EP_IPC_POLL_EVENTS_HANGUP;
+				poll_handles_data [index].events = (uint8_t)DS_IPC_POLL_EVENTS_HANGUP;
 			} else {
 				if (callback)
 					callback ("Client connection error", error);
-				poll_handles_data [index].events = (uint8_t)EP_IPC_POLL_EVENTS_ERR;
+				poll_handles_data [index].events = (uint8_t)DS_IPC_POLL_EVENTS_ERR;
 				result = -1;
 				ep_raise_error ();
 			}
 		} else {
-			poll_handles_data [index].events = (uint8_t)EP_IPC_POLL_EVENTS_SIGNALED;
+			poll_handles_data [index].events = (uint8_t)DS_IPC_POLL_EVENTS_SIGNALED;
 		}
 	} else {
 		// SERVER
-		poll_handles_data [index].events = (uint8_t)EP_IPC_POLL_EVENTS_SIGNALED;
+		poll_handles_data [index].events = (uint8_t)DS_IPC_POLL_EVENTS_SIGNALED;
 	}
 
 	result = 1;
@@ -835,13 +835,14 @@ ipc_stream_close_func (void *object)
 }
 
 static
-EventPipeIpcPollEvents
+DiagnosticsIpcPollEvents
 ipc_stream_poll_func (
 	void *object,
 	uint32_t timeout_ms)
 {
-	// Needs to be implemented.
-	return EP_IPC_POLL_EVENTS_UNKNOWN;
+	EP_UNREACHABLE ("ipc_stream_poll_func needs to be implemented for NamedPipes");
+	// TODO: Implement ipc_stream_poll_func for NamedPipes
+	return DS_IPC_POLL_EVENTS_UNKNOWN;
 }
 
 static IpcStreamVtable ipc_stream_vtable = {

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)EP_IPC_POLL_EVENTS_HANGUP;
+				poll_handles_data [i].events = (uint8_t)DS_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)EP_IPC_POLL_EVENTS_ERR;
+				poll_handles_data [i].events = (uint8_t)DS_IPC_POLL_EVENTS_ERR;
 			} else if (poll_fds [i].revents & (POLLIN|POLLPRI)) {
-				poll_handles_data [i].events = (uint8_t)EP_IPC_POLL_EVENTS_SIGNALED;
+				poll_handles_data [i].events = (uint8_t)DS_IPC_POLL_EVENTS_SIGNALED;
 			} else {
-				poll_handles_data [i].events = (uint8_t)EP_IPC_POLL_EVENTS_UNKNOWN;
+				poll_handles_data [i].events = (uint8_t)DS_IPC_POLL_EVENTS_UNKNOWN;
 				if (callback)
 					callback ("unknown poll response", (uint32_t)poll_fds [i].revents);
 			}
@@ -1490,7 +1490,7 @@ ipc_stream_close_func (void *object)
 }
 
 static
-EventPipeIpcPollEvents
+DiagnosticsIpcPollEvents
 ipc_stream_poll_func (
 	void *object,
 	uint32_t timeout_ms)
@@ -1680,15 +1680,15 @@ ds_ipc_stream_to_string (
 	return (result > 0 && result < (int32_t)buffer_len) ? result : 0;
 }
 
-EventPipeIpcPollEvents
+DiagnosticsIpcPollEvents
 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;
+		return DS_IPC_POLL_EVENTS_HANGUP;
 
 	ds_ipc_pollfd_t pfd;
 	pfd.fd = ipc_stream->client_socket;
@@ -1698,24 +1698,24 @@ ds_ipc_stream_poll (
 	result_poll = ipc_poll_fds (&pfd, 1, timeout_ms);
 
 	if (result_poll < 0)
-		return EP_IPC_POLL_EVENTS_ERR;
+		return DS_IPC_POLL_EVENTS_ERR;
 
 	if (result_poll == 0)
-		return EP_IPC_POLL_EVENTS_HANGUP;
+		return DS_IPC_POLL_EVENTS_NONE;
 
 	if (pfd.revents == 0)
-		return EP_IPC_POLL_EVENTS_NONE;
+		return DS_IPC_POLL_EVENTS_NONE;
 
 	if (pfd.revents & POLLHUP)
-		return EP_IPC_POLL_EVENTS_HANGUP;
+		return DS_IPC_POLL_EVENTS_HANGUP;
 
 	if (pfd.revents & (POLLERR | POLLNVAL))
-		return EP_IPC_POLL_EVENTS_ERR;
+		return DS_IPC_POLL_EVENTS_ERR;
 
 	if (pfd.revents & (POLLIN | POLLPRI | POLLOUT))
-		return EP_IPC_POLL_EVENTS_SIGNALED;
+		return DS_IPC_POLL_EVENTS_SIGNALED;
 
-	return EP_IPC_POLL_EVENTS_UNKNOWN;
+	return DS_IPC_POLL_EVENTS_UNKNOWN;
 }
 
 #endif /* ENABLE_PERFTRACING */

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

@@ -22,6 +22,14 @@ 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)EP_IPC_POLL_EVENTS_ERR;
+			poll_handles_data [i].events = (uint8_t)DS_IPC_POLL_EVENTS_ERR;
 			return 1;
 		}
 		if (pending > 0){
-			poll_handles_data [i].events = (uint8_t)EP_IPC_POLL_EVENTS_SIGNALED;
+			poll_handles_data [i].events = (uint8_t)DS_IPC_POLL_EVENTS_SIGNALED;
 			return 1;
 		}
 	}
@@ -426,23 +426,14 @@ ipc_stream_close_func (void *object)
 }
 
 static
-EventPipeIpcPollEvents
+DiagnosticsIpcPollEvents
 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;
+	EP_UNREACHABLE ("ipc_stream_poll_func needs to be implemented for WebSockets");
+	// TODO: Implement ipc_stream_poll_func for WebSockets
+	return DS_IPC_POLL_EVENTS_UNKNOWN;
 }
 
 static IpcStreamVtable ipc_stream_vtable = {

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

@@ -140,7 +140,7 @@ ds_ipc_stream_to_string (
 	ep_char8_t *buffer,
 	uint32_t buffer_len);
 
-EventPipeIpcPollEvents
+DiagnosticsIpcPollEvents
 ds_ipc_stream_poll (
 	DiagnosticsIpcStream *ipc_stream,
 	uint32_t timeout_ms);

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 EP_IPC_POLL_EVENTS_HANGUP:
+				case DS_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 EP_IPC_POLL_EVENTS_SIGNALED:
+				case DS_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 EP_IPC_POLL_EVENTS_ERR:
+				case DS_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 EP_IPC_POLL_EVENTS_NONE:
+				case DS_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,19 +11,5 @@
 
 #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,7 +21,7 @@ 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);
+typedef DiagnosticsIpcPollEvents (*IpcStreamPollFunc)(void *object, uint32_t timeout_ms);
 
 struct _IpcStreamVtable {
 	IpcStreamFreeFunc free_func;
@@ -79,7 +79,7 @@ ep_ipc_stream_flush_vcall (IpcStream *ipc_stream);
 bool
 ep_ipc_stream_close_vcall (IpcStream *ipc_stream);
 
-EventPipeIpcPollEvents
+DiagnosticsIpcPollEvents
 ep_ipc_stream_poll_vcall (IpcStream *ipc_stream, uint32_t timeout_ms);
 
 #endif /* ENABLE_PERFTRACING */

src/native/eventpipe/ep-session.c

@@ -76,6 +76,10 @@ session_tracepoint_write_event (
 	ep_rt_thread_handle_t event_thread,
 	EventPipeStackContents *stack);
 
+static
+bool
+session_is_stream_connection_closed (IpcStream *stream);
+
 /*
  * EventPipeSession.
  */
@@ -123,15 +127,15 @@ EP_RT_DEFINE_THREAD_FUNC (streaming_thread)
 				ep_rt_thread_sleep (timeout_ns);
 			}
 		} else if (session->session_type == EP_SESSION_TYPE_USEREVENTS) {
-			// User events session, write all user events tracepoints to the file.
+			// In a user events session we only monitor the stream to stop the session if it closes.
 			while (ep_session_get_streaming_enabled (session)) {
-				EP_ASSERT (session->continuation_stream != NULL);
-				if (ep_ipc_continuation_stream_connection_closed (session->continuation_stream)) {
+				EP_ASSERT (session->stream != NULL);
+				if (session_is_stream_connection_closed (session->stream)) {
 					success = false;
 					break;
 				}
 
-				// Wait until it's time to sample again.
+				// Wait until it's time to poll again.
 				const uint32_t timeout_ns = 100000000; // 100 msec.
 				ep_rt_thread_sleep (timeout_ns);
 			}
@@ -219,6 +223,15 @@ session_disable_streaming_thread (EventPipeSession *session)
 	ep_rt_wait_event_free (rt_thread_shutdown_event);
 }
 
+static
+bool
+session_is_stream_connection_closed (IpcStream *stream)
+{
+	EP_ASSERT (stream != NULL);
+	DiagnosticsIpcPollEvents poll_event = ep_ipc_stream_poll_vcall (stream, DS_IPC_TIMEOUT_INFINITE);
+	return poll_event == DS_IPC_POLL_EVENTS_HANGUP;
+}
+
 /*
  *  session_user_events_tracepoints_init
  *
@@ -300,7 +313,7 @@ ep_session_alloc (
 	instance->synchronous_callback = sync_callback;
 	instance->callback_additional_data = callback_additional_data;
 	instance->user_events_data_fd = -1;
-	instance->continuation_stream = NULL;
+	instance->stream = NULL;
 
 	// Hard coded 10MB for now, we'll probably want to make
 	// this configurable later.
@@ -339,7 +352,7 @@ ep_session_alloc (
 	case EP_SESSION_TYPE_USEREVENTS:
 		// With the user_events_data file, register tracepoints for each provider's tracepoint configurations
 		ep_raise_error_if_nok (session_user_events_tracepoints_init (instance, user_events_data_fd));
-		instance->continuation_stream = (IpcContinuationStream *)stream;
+		instance->stream = stream;
 		break;
 
 	default:

src/native/eventpipe/ep-session.h

@@ -72,7 +72,8 @@ struct _EventPipeSession_Internal {
 	// The user_events_data file descriptor to register Tracepoints and write user_events to.
 	int user_events_data_fd;
 	// The IPC continuation stream from initializing the session through the diagnostic server
-	IpcContinuationStream *continuation_stream;
+	// Currently only initialized for user_events sessions.
+	IpcStream *stream;
 };
 
 #if !defined(EP_INLINE_GETTER_SETTER) && !defined(EP_IMPL_SESSION_GETTER_SETTER)