X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=ustd%2Fustd.c;h=4db550222407ba7d0773aa8b15e1b33cda777de2;hb=2a79ceebcba229c91c93272705444c61c458bb06;hp=969c192abdb66bea7b9ce93dd48d7a994b5fe70c;hpb=204141ee9da22a244c9095287f4f1c513784b171;p=ust.git

diff --git a/ustd/ustd.c b/ustd/ustd.c
index 969c192..4db5502 100644
--- a/ustd/ustd.c
+++ b/ustd/ustd.c
@@ -48,6 +48,7 @@
 #define PUT_SUBBUF_OK 1
 #define PUT_SUBBUF_DIED 0
 #define PUT_SUBBUF_PUSHED 2
+#define PUT_SUBBUF_DONE 3
 
 char *sock_path=NULL;
 char *trace_path=NULL;
@@ -98,6 +99,11 @@ int get_subbuffer(struct buffer_info *buf)
 		retval = GET_SUBBUF_DONE;
 		goto end_rep;
 	}
+	else if(!strcmp(received_msg, "NOTFOUND")) {
+		WARN("For buffer %s, the trace was not found. This likely means it was destroyed by the user.", buf->name);
+		retval = GET_SUBBUF_DONE;
+		goto end_rep;
+	}
 	else {
 		DBG("error getting subbuffer %s", buf->name);
 		retval = -1;
@@ -130,11 +136,18 @@ int put_subbuffer(struct buffer_info *buf)
 		retval = PUT_SUBBUF_DIED;
 		goto end;
 	}
-	if(result < 0) {
+	else if(result < 0) {
 		ERR("put_subbuffer: send_message failed");
 		retval = -1;
 		goto end;
 	}
+	else if(result == 0) {
+		/* Program seems finished. However this might not be
+		 * the last subbuffer that has to be collected.
+		 */
+		retval = PUT_SUBBUF_DIED;
+		goto end;
+	}
 
 	result = sscanf(received_msg, "%as", &rep_code);
 	if(result != 1) {
@@ -147,6 +160,14 @@ int put_subbuffer(struct buffer_info *buf)
 		DBG("subbuffer put %s", buf->name);
 		retval = PUT_SUBBUF_OK;
 	}
+	else if(!strcmp(received_msg, "NOTFOUND")) {
+		WARN("For buffer %s, the trace was not found. This likely means it was destroyed by the user.", buf->name);
+		/* However, maybe this was not the last subbuffer. So
+		 * we return the program died.
+		 */
+		retval = PUT_SUBBUF_DIED;
+		goto end_rep;
+	}
 	else {
 		DBG("put_subbuffer: received error, we were pushed");
 		retval = PUT_SUBBUF_PUSHED;
@@ -239,6 +260,9 @@ struct buffer_info *connect_buffer(pid_t pid, const char *bufname)
 		ERR("problem in ustcomm_send_request(get_pidunique)");
 		return NULL;
 	}
+	if(result == 0) {
+		goto error;
+	}
 
 	result = sscanf(received_msg, "%lld", &buf->pidunique);
 	if(result != 1) {
@@ -256,6 +280,9 @@ struct buffer_info *connect_buffer(pid_t pid, const char *bufname)
 		ERR("problem in ustcomm_send_request(get_shmid)");
 		return NULL;
 	}
+	if(result == 0) {
+		goto error;
+	}
 
 	result = sscanf(received_msg, "%d %d", &buf->shmid, &buf->bufstruct_shmid);
 	if(result != 2) {
@@ -273,6 +300,9 @@ struct buffer_info *connect_buffer(pid_t pid, const char *bufname)
 		ERR("problem in ustcomm_send_request(g_n_subbufs)");
 		return NULL;
 	}
+	if(result == 0) {
+		goto error;
+	}
 
 	result = sscanf(received_msg, "%d", &buf->n_subbufs);
 	if(result != 1) {
@@ -284,8 +314,15 @@ struct buffer_info *connect_buffer(pid_t pid, const char *bufname)
 
 	/* get subbuf size */
 	asprintf(&send_msg, "get_subbuf_size %s", buf->name);
-	ustcomm_send_request(&buf->conn, send_msg, &received_msg);
+	result = ustcomm_send_request(&buf->conn, send_msg, &received_msg);
 	free(send_msg);
+	if(result == -1) {
+		ERR("problem in ustcomm_send_request(get_subbuf_size)");
+		return NULL;
+	}
+	if(result == 0) {
+		goto error;
+	}
 
 	result = sscanf(received_msg, "%d", &buf->subbuf_size);
 	if(result != 1) {
@@ -357,6 +394,56 @@ struct buffer_info *connect_buffer(pid_t pid, const char *bufname)
 	pthread_mutex_unlock(&active_buffers_mutex);
 
 	return buf;
+
+error:
+	free(buf);
+	return NULL;
+}
+
+int unwrite_last_subbuffer(struct buffer_info *buf)
+{
+	int result;
+
+	result = ftruncate(buf->file_fd, buf->previous_offset);
+	if(result == -1) {
+		PERROR("ftruncate");
+		return -1;
+	}
+
+	result = lseek(buf->file_fd, buf->previous_offset, SEEK_SET);
+	if(result == (int)(off_t)-1) {
+		PERROR("lseek");
+		return -1;
+	}
+
+	return 0;
+}
+
+int write_current_subbuffer(struct buffer_info *buf)
+{
+	int result;
+
+	void *subbuf_mem = buf->mem + (buf->consumed_old & (buf->n_subbufs * buf->subbuf_size-1));
+
+	size_t cur_sb_size = subbuffer_data_size(subbuf_mem);
+
+	off_t cur_offset = lseek(buf->file_fd, 0, SEEK_CUR);
+	if(cur_offset == (off_t)-1) {
+		PERROR("lseek");
+		return -1;
+	}
+
+	buf->previous_offset = cur_offset;
+	DBG("previous_offset: %ld", cur_offset);
+
+	result = patient_write(buf->file_fd, subbuf_mem, cur_sb_size);
+	if(result == -1) {
+		PERROR("write");
+		/* FIXME: maybe drop this trace */
+		return -1;
+	}
+
+	return 0;
 }
 
 int consumer_loop(struct buffer_info *buf)
@@ -382,14 +469,10 @@ int consumer_loop(struct buffer_info *buf)
 		}
 
 		/* write data to file */
-		result = patient_write(buf->file_fd, buf->mem + (buf->consumed_old & (buf->n_subbufs * buf->subbuf_size-1)), buf->subbuf_size);
-		if(result == -1) {
-			PERROR("write");
-			/* FIXME: maybe drop this trace */
-		}
+		write_current_subbuffer(buf);
+		/* FIXME: handle return value? */
 
 		/* put the subbuffer */
-		/* FIXME: we actually should unput the buffer before consuming... */
 		result = put_subbuffer(buf);
 		if(result == -1) {
 			ERR("unknown error putting subbuffer (channel=%s)", buf->name);
@@ -401,10 +484,22 @@ int consumer_loop(struct buffer_info *buf)
 		}
 		else if(result == PUT_SUBBUF_DIED) {
 			WARN("application died while putting subbuffer");
-			/* FIXME: probably need to skip the first subbuffer in finish_consuming_dead_subbuffer */
+			/* Skip the first subbuffer. We are not sure it is trustable
+			 * because the put_subbuffer() did not complete.
+			 */
+			unwrite_last_subbuffer(buf);
 			finish_consuming_dead_subbuffer(buf);
 			break;
 		}
+		else if(result == PUT_SUBBUF_DONE) {
+			/* Done with this subbuffer */
+			/* FIXME: add a case where this branch is used? Upon
+			 * normal trace termination, at put_subbuf time, a
+			 * special last-subbuffer code could be returned by
+			 * the listener.
+			 */
+			break;
+		}
 		else if(result == PUT_SUBBUF_OK) {
 		}
 	}
@@ -579,6 +674,11 @@ int start_ustd(int fd)
 		PERROR("sigaction");
 		return 1;
 	}
+	result = sigaction(SIGINT, &sa, NULL);
+	if(result == -1) {
+		PERROR("sigaction");
+		return 1;
+	}
 
 	result = ustcomm_init_ustd(&ustd, sock_path);
 	if(result == -1) {
@@ -661,6 +761,9 @@ int start_ustd(int fd)
 				free_bufname:
 				free(bufname);
 			}
+			else {
+				WARN("unknown command: %s", recvbuf);
+			}
 
 			free(recvbuf);
 		}
@@ -677,6 +780,8 @@ int start_ustd(int fd)
 		}
 	}
 
+	ustcomm_fini_ustd(&ustd);
+
 	return 0;
 }