X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=liblttkconsumerd%2Fliblttkconsumerd.c;h=e11229abd5dca98661aa653be113c6b74a6e088c;hp=4ec949e0f2c18ad771f6ad3e30151e4a0af28bf2;hb=4de84ad900e5b6e739f3c36aa1b86746081333d7;hpb=fec07047f5911e611d7b0d0edf082f9c1b013cdb

diff --git a/liblttkconsumerd/liblttkconsumerd.c b/liblttkconsumerd/liblttkconsumerd.c
index 4ec949e0f..e11229abd 100644
--- a/liblttkconsumerd/liblttkconsumerd.c
+++ b/liblttkconsumerd/liblttkconsumerd.c
@@ -4,8 +4,8 @@
  *
  * This program is free software; you can redistribute it and/or
  * modify it under the terms of the GNU General Public License
- * as published by the Free Software Foundation; either version 2
- * of the License, or (at your option) any later version.
+ * as published by the Free Software Foundation; only version 2
+ * of the License.
  *
  * This program is distributed in the hope that it will be useful,
  * but WITHOUT ANY WARRANTY; without even the implied warranty of
@@ -69,12 +69,8 @@ static int kconsumerd_thread_pipe[2];
 /* pipe to wake the poll thread when necessary */
 static int kconsumerd_poll_pipe[2];
 
-/*
- * TODO: create a should_quit pipe to let the signal handler wake up the
- * fd receiver thread. It should be initialized before any signal can be
- * received by the library.
- */
-
+/* to let the signal handler wake up the fd receiver thread */
+static int kconsumerd_should_quit[2];
 
 /* timeout parameter, to control the polling thread grace period */
 static int kconsumerd_poll_timeout = -1;
@@ -495,7 +491,7 @@ static int kconsumerd_read_subbuffer(struct kconsumerd_fd *kconsumerd_fd)
 	}
 
 	switch (DEFAULT_KERNEL_CHANNEL_OUTPUT) {
-	case LTTNG_KERNEL_SPLICE:
+	case LTTNG_EVENT_SPLICE:
 		/* read the whole subbuffer */
 		err = kernctl_get_padded_subbuf_size(infd, &len);
 		if (err != 0) {
@@ -514,7 +510,7 @@ static int kconsumerd_read_subbuffer(struct kconsumerd_fd *kconsumerd_fd)
 			ERR("Error splicing to tracefile");
 		}
 		break;
-	case LTTNG_KERNEL_MMAP:
+	case LTTNG_EVENT_MMAP:
 		/* read the used subbuffer size */
 		err = kernctl_get_subbuf_size(infd, &len);
 		if (err != 0) {
@@ -553,6 +549,32 @@ end:
 	return ret;
 }
 
+/*
+ * kconsumerd_poll_socket
+ *
+ * Poll on the should_quit pipe and the command socket
+ * return -1 on error and should exit, 0 if data is
+ * available on the command socket
+ */
+int kconsumerd_poll_socket(struct pollfd *kconsumerd_sockpoll)
+{
+	int num_rdy;
+
+	num_rdy = poll(kconsumerd_sockpoll, 2, -1);
+	if (num_rdy == -1) {
+		perror("Poll error");
+		goto exit;
+	}
+	if (kconsumerd_sockpoll[0].revents == POLLIN) {
+		DBG("kconsumerd_should_quit wake up");
+		goto exit;
+	}
+	return 0;
+
+exit:
+	return -1;
+}
+
 /*
  * kconsumerd_consumerd_recv_fd
  *
@@ -560,7 +582,8 @@ end:
  * structures describing each fd (path name).
  * Returns the size of received data
  */
-static int kconsumerd_consumerd_recv_fd(int sfd, int size,
+static int kconsumerd_consumerd_recv_fd(int sfd,
+		struct pollfd *kconsumerd_sockpoll, int size,
 		enum kconsumerd_command cmd_type)
 {
 	struct msghdr msg;
@@ -587,6 +610,10 @@ static int kconsumerd_consumerd_recv_fd(int sfd, int size,
 		msg.msg_controllen = sizeof(recv_fd);
 
 		DBG("Waiting to receive fd");
+		if (kconsumerd_poll_socket(kconsumerd_sockpoll) < 0) {
+			goto end;
+		}
+
 		if ((ret = recvmsg(sfd, &msg, 0)) < 0) {
 			perror("recvmsg");
 			continue;
@@ -805,13 +832,34 @@ end:
 }
 
 /*
- * kconsumerd_create_poll_pipe
+ * kconsumerd_init(void)
  *
- * create the pipe to wake to polling thread when needed
+ * initialise the necessary environnement :
+ * - inform the polling thread to update the polling array
+ * - create the poll_pipe
+ * - create the should_quit pipe (for signal handler)
  */
-int kconsumerd_create_poll_pipe()
+int kconsumerd_init(void)
 {
-	return pipe(kconsumerd_poll_pipe);
+	int ret;
+
+	/* need to update the polling array at init time */
+	kconsumerd_data.need_update = 1;
+
+	ret = pipe(kconsumerd_poll_pipe);
+	if (ret < 0) {
+		perror("Error creating poll pipe");
+		goto end;
+	}
+
+	ret = pipe(kconsumerd_should_quit);
+	if (ret < 0) {
+		perror("Error creating recv pipe");
+		goto end;
+	}
+
+end:
+	return ret;
 }
 
 /*
@@ -824,6 +872,12 @@ void *kconsumerd_thread_receive_fds(void *data)
 {
 	int sock, client_socket, ret;
 	struct lttcomm_kconsumerd_header tmp;
+	/*
+	 * structure to poll for incoming data on communication socket
+	 * avoids making blocking sockets
+	 */
+	struct pollfd kconsumerd_sockpoll[2];
+
 
 	DBG("Creating command socket %s", kconsumerd_command_sock_path);
 	unlink(kconsumerd_command_sock_path);
@@ -845,18 +899,46 @@ void *kconsumerd_thread_receive_fds(void *data)
 		goto end;
 	}
 
-	/* TODO: poll on socket and "should_quit" fd pipe */
-	/* TODO: change blocking call into non-blocking call */
+	ret = fcntl(client_socket, F_SETFL, O_NONBLOCK);
+	if (ret < 0) {
+		perror("fcntl O_NONBLOCK");
+		goto end;
+	}
+
+	/* prepare the FDs to poll : to client socket and the should_quit pipe */
+	kconsumerd_sockpoll[0].fd = kconsumerd_should_quit[0];
+	kconsumerd_sockpoll[0].events = POLLIN | POLLPRI;
+	kconsumerd_sockpoll[1].fd = client_socket;
+	kconsumerd_sockpoll[1].events = POLLIN | POLLPRI;
+
+	if (kconsumerd_poll_socket(kconsumerd_sockpoll) < 0) {
+		goto end;
+	}
+	DBG("Connection on client_socket");
+
 	/* Blocking call, waiting for transmission */
 	sock = lttcomm_accept_unix_sock(client_socket);
 	if (sock <= 0) {
 		WARN("On accept");
 		goto end;
 	}
+	ret = fcntl(sock, F_SETFL, O_NONBLOCK);
+	if (ret < 0) {
+		perror("fcntl O_NONBLOCK");
+		goto end;
+	}
+
+	/* update the polling structure to poll on the established socket */
+	kconsumerd_sockpoll[1].fd = sock;
+	kconsumerd_sockpoll[1].events = POLLIN | POLLPRI;
+
 	while (1) {
+		if (kconsumerd_poll_socket(kconsumerd_sockpoll) < 0) {
+			goto end;
+		}
+		DBG("Incoming fds on sock");
+
 		/* We first get the number of fd we are about to receive */
-		/* TODO: poll on sock and "should_quit" fd pipe */
-		/* TODO: change recv into a non-blocking call */
 		ret = lttcomm_recv_unix_sock(sock, &tmp,
 				sizeof(struct lttcomm_kconsumerd_header));
 		if (ret <= 0) {
@@ -871,12 +953,15 @@ void *kconsumerd_thread_receive_fds(void *data)
 			DBG("kconsumerd_thread_receive_fds received quit from signal");
 			goto end;
 		}
+
 		/* we received a command to add or update fds */
-		ret = kconsumerd_consumerd_recv_fd(sock, tmp.payload_size, tmp.cmd_type);
+		ret = kconsumerd_consumerd_recv_fd(sock, kconsumerd_sockpoll,
+				tmp.payload_size, tmp.cmd_type);
 		if (ret <= 0) {
 			ERR("Receiving the FD, exiting");
 			goto end;
 		}
+		DBG("received fds on sock");
 	}
 
 end:
@@ -926,15 +1011,15 @@ void kconsumerd_cleanup(void)
 }
 
 /*
+ * kconsumerd_should_exit
+ *
  * Called from signal handler.
  */
 void kconsumerd_should_exit(void)
 {
+	int ret;
 	kconsumerd_quit = 1;
-	/*
-	 * TODO: write into a should_quit pipe to wake up the fd
-	 * receiver thread.
-	 */
+	ret = write(kconsumerd_should_quit[1], "4", 1);
 }
 
 /*