fix(java): improve socket lifecycle handling for connect/send/close (#11760)

* fix(eBPF): Use MSG_NOSIGNAL in send() to avoid SIGPIPE on Linux

* fix(java): serialize socket fd close and send paths

Split close_files() into a locked wrapper and a non-locking close_files_locked() helper to ensure fd close operations share the same mutex with send paths. Update send_msg() to use close_files_locked() on send failure to avoid recursive locking when already holding g_df_lock. Move perf_map_socket_fd validity checks in df_send_symbol() inside the critical section to eliminate races between fd close and send operations during re-attach or error handling. This change only tightens fd lifecycle synchronization and does not alter JVMTI event lifecycle or replay semantics.

* Add `close()` when `connect()` fails.

* Adjust comments

Author: yinjiping

agent/src/ebpf/Makefile

@@ -102,7 +102,7 @@ OBJS := user/elf.o \
 	user/profile/java/collect_symbol_files.o
 
 JAVA_TOOL := deepflow-jattach
-JAVA_AGENT_VERSION := 4
+JAVA_AGENT_VERSION := 4.1
 JAVA_AGENT_GNU_SO := df_java_agent_v$(JAVA_AGENT_VERSION).so
 JAVA_AGENT_MUSL_SO := df_java_agent_musl_v$(JAVA_AGENT_VERSION).so
 JAVA_AGENT_SO := $(JAVA_AGENT_GNU_SO) $(JAVA_AGENT_MUSL_SO)

agent/src/ebpf/user/profile/java/symbol_collect_agent.c

@@ -68,6 +68,7 @@ int perf_map_log_socket_fd = -1;
 // Cache symbols for batch sending
 char g_symbol_buffer[STRING_BUFFER_SIZE * 4];
 int g_cached_bytes;
+static jint close_files_locked(void);
 jint close_files(void);
 
 #define _(e)                                                                \
@@ -94,19 +95,23 @@ inline int send_msg(int sock_fd, const char *buf, size_t len)
 	int n = 0;		// Initialize n
 
 	do {
-		n = send(sock_fd, buf + send_bytes, len - send_bytes, 0);
+		/*
+		 * Note: To avoid SIGPIPE signal (which terminates the process), use
+		 * MSG_NOSIGNAL flag in send() call.
+		 */
+		n = send(sock_fd, buf + send_bytes, len - send_bytes, MSG_NOSIGNAL);
 		if (n == -1) {
 			if (errno == EINTR || errno == EAGAIN
 			    || errno == EWOULDBLOCK) {
 				// Retry on interrupt or temporary failure
 				continue;
 			} else {
-				close_files();	// Example function call, define as needed
+				close_files_locked();	// Example function call, define as needed
 				break;
 			}
 		} else if (n == 0) {
 			// Connection closed by peer
-			close_files();	// Example function call, define as needed
+			close_files_locked();	// Example function call, define as needed
 			break;
 		}
 
@@ -126,11 +131,9 @@ jint df_open_socket(const char *path, int *ptr)
 
 	/*
 	 * The reason for setting non-blocking mode:
-	 * 1 To prevent Java threads from being blocked.
-	 * 2 When attempts to write data to a closed writing port of a pipe or
-	 *   socket, the operating system detects this situation and sends the
-	 *   SIGPIPE signal to Java process, which causes the program to exit.
-	 *   Use non-blocking mode to avoid this issue.
+	 * 1 To prevent Java threads from being blocked when writing to socket.
+	 * 2 Non-blocking mode allows send() to fail with EAGAIN/EWOULDBLOCK
+	 *   instead of blocking, enabling graceful error handling.
 	 */
 	int flags = fcntl(s, F_GETFL, 0);
 	if (flags == -1) {
@@ -148,6 +151,7 @@ jint df_open_socket(const char *path, int *ptr)
 	strncpy(remote.sun_path, path, UNIX_PATH_MAX - 1);
 	int len = sizeof(remote.sun_family) + strlen(remote.sun_path);
 	if (connect(s, (struct sockaddr *)&remote, len) == -1) {
+		close(s);
 		fprintf(stderr, "Call connect() failed: errno(%d)\n", errno);
 		return JNI_ERR;
 	}
@@ -215,21 +219,32 @@ jint df_agent_config(char *opts)
 	return JNI_OK;
 }
 
-jint close_files(void)
+static jint close_files_locked(void)
 {
-	if (perf_map_socket_fd > 0) {
-		close(perf_map_socket_fd);
-		perf_map_socket_fd = -1;
-	}
+	int perf_fd = perf_map_socket_fd;
+	int log_fd = perf_map_log_socket_fd;
+
+	perf_map_socket_fd = -1;
+	perf_map_log_socket_fd = -1;
 
-	if (perf_map_log_socket_fd > 0) {
-		close(perf_map_log_socket_fd);
-		perf_map_log_socket_fd = -1;
+	if (perf_fd > 0) {
+		close(perf_fd);
+	}
+	if (log_fd > 0) {
+		close(log_fd);
 	}
 
 	return JNI_OK;
 }
 
+jint close_files(void)
+{
+	pthread_mutex_lock(&g_df_lock);
+	close_files_locked();
+	pthread_mutex_unlock(&g_df_lock);
+	return JNI_OK;
+}
+
 JNIEXPORT uint64_t df_java_agent_so_libs_test(void)
 {
 	/*
@@ -302,13 +317,16 @@ void deallocate(jvmtiEnv * jvmti, void *string)
 void df_send_symbol(enum event_type type, const void *code_addr,
 		    unsigned int code_size, const char *entry)
 {
+	int send_bytes;
+	struct symbol_metadata *meta;
+	char symbol_str[STRING_BUFFER_SIZE];
+
+	pthread_mutex_lock(&g_df_lock);
 	if (perf_map_socket_fd < 0) {
+		pthread_mutex_unlock(&g_df_lock);
 		return;
 	}
 
-	int send_bytes;
-	struct symbol_metadata *meta;
-	char symbol_str[STRING_BUFFER_SIZE];
 	if (type == METHOD_UNLOAD) {
 		snprintf(symbol_str + sizeof(*meta),
 			 sizeof(symbol_str) - sizeof(*meta), "%lx",
@@ -322,7 +340,6 @@ void df_send_symbol(enum event_type type, const void *code_addr,
 	meta->len = strlen(symbol_str + sizeof(*meta));
 	meta->type = type;
 	send_bytes = meta->len + sizeof(*meta);
-	pthread_mutex_lock(&g_df_lock);
 	if (replay_finish) {
 		send_msg(perf_map_socket_fd, symbol_str, send_bytes);
 	} else {