*
* 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
#include <sys/types.h>
#include <unistd.h>
#include <urcu/list.h>
+#include <assert.h>
#include "libkernelctl.h"
#include "liblttkconsumerd.h"
/* 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;
* kconsumerd_find_session_fd
*
* Find a session fd in the global list.
+ * The kconsumerd_data.lock must be locked during this call
*
* Return 1 if found else 0
*/
{
struct kconsumerd_fd *iter;
- pthread_mutex_lock(&kconsumerd_data.lock);
cds_list_for_each_entry(iter, &kconsumerd_data.fd_list.head, list) {
if (iter->sessiond_fd == fd) {
DBG("Duplicate session fd %d", fd);
- pthread_mutex_unlock(&kconsumerd_data.lock);
return 1;
}
}
- pthread_mutex_unlock(&kconsumerd_data.lock);
return 0;
}
tmp_fd->state = buf->state;
tmp_fd->max_sb_size = buf->max_sb_size;
strncpy(tmp_fd->path_name, buf->path_name, PATH_MAX);
+ tmp_fd->path_name[PATH_MAX - 1] = '\0';
/* Opening the tracefile in write mode */
ret = open(tmp_fd->path_name,
}
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) {
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) {
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
*
* 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;
struct iovec iov[1];
int ret = 0, i, tmp2;
struct cmsghdr *cmsg;
/* the number of fds we are about to receive */
nb_fd = size / sizeof(struct lttcomm_kconsumerd_msg);
+ /*
+ * nb_fd is the number of fds we receive. One fd per recvmsg.
+ */
for (i = 0; i < nb_fd; i++) {
- memset(&msg, 0, sizeof(msg));
+ struct msghdr msg = { 0 };
/* Prepare to receive the structures */
iov[0].iov_base = &lkm;
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;
kconsumerd_send_error(KCONSUMERD_ERROR_RECV_FD);
goto end;
}
+
/* if we received fds */
if (cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_RIGHTS) {
switch (cmd_type) {
case ADD_STREAM:
- DBG("kconsumerd_add_fd %s (%d)", lkm.path_name, (CMSG_DATA(cmsg)[0]));
- ret = kconsumerd_add_fd(&lkm, (CMSG_DATA(cmsg)[0]));
+ DBG("kconsumerd_add_fd %s (%d)", lkm.path_name, ((int *) CMSG_DATA(cmsg))[0]);
+ ret = kconsumerd_add_fd(&lkm, ((int *) CMSG_DATA(cmsg))[0]);
if (ret < 0) {
kconsumerd_send_error(KCONSUMERD_OUTFD_ERROR);
goto end;
}
/* signal the poll thread */
tmp2 = write(kconsumerd_poll_pipe[1], "4", 1);
+ if (tmp2 < 0) {
+ perror("write kconsumerd poll");
+ }
} else {
ERR("Didn't received any fd");
kconsumerd_send_error(KCONSUMERD_ERROR_RECV_FD);
if (pollfd[nb_fd].revents == POLLIN) {
DBG("kconsumerd_poll_pipe wake up");
tmp2 = read(kconsumerd_poll_pipe[0], &tmp, 1);
+ if (tmp2 < 0) {
+ perror("read kconsumerd poll");
+ }
continue;
}
}
/*
- * 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;
}
/*
{
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);
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) {
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:
*/
void kconsumerd_cleanup(void)
{
- struct kconsumerd_fd *iter;
+ struct kconsumerd_fd *iter, *tmp;
/* remove the socket file */
unlink(kconsumerd_command_sock_path);
* running (after joining on the threads), no need to protect
* list iteration with mutex.
*/
- cds_list_for_each_entry(iter, &kconsumerd_data.fd_list.head, list) {
+ cds_list_for_each_entry_safe(iter, tmp, &kconsumerd_data.fd_list.head, list) {
kconsumerd_del_fd(iter);
}
}
/*
+ * 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);
+ if (ret < 0) {
+ perror("write kconsumerd quit");
+ }
}
/*
projects / lttng-tools.git / blobdiff