#include "liblttkconsumerd.h"
#include "lttngerr.h"
-/* Init the list of FDs */
-static struct kconsumerd_fd_list kconsumerd_fd_list = {
- .head = CDS_LIST_HEAD_INIT(kconsumerd_fd_list.head),
+static
+struct kconsumerd_global_data {
+ /*
+ * kconsumerd_data.lock protects kconsumerd_data.fd_list,
+ * kconsumerd_data.fds_count, and kconsumerd_data.need_update. It
+ * ensures the count matches the number of items in the fd_list.
+ * It ensures the list updates *always* trigger an fd_array
+ * update (therefore need to make list update vs
+ * kconsumerd_data.need_update flag update atomic, and also flag
+ * read, fd array and flag clear atomic).
+ */
+ pthread_mutex_t lock;
+ /*
+ * Number of element for the list below. Protected by
+ * kconsumerd_data.lock.
+ */
+ unsigned int fds_count;
+ /*
+ * List of FDs. Protected by kconsumerd_data.lock.
+ */
+ struct kconsumerd_fd_list fd_list;
+ /*
+ * Flag specifying if the local array of FDs needs update in the
+ * poll function. Protected by kconsumerd_data.lock.
+ */
+ unsigned int need_update;
+} kconsumerd_data = {
+ .fd_list.head = CDS_LIST_HEAD_INIT(kconsumerd_data.fd_list.head),
};
-/* Number of element for the list below. */
-static unsigned int kconsumerd_fds_count;
-
-/* If the local array of FDs needs update in the poll function */
-static unsigned int kconsumerd_update_fd_array = 1;
-
-/* lock the fd array and structures */
-static pthread_mutex_t kconsumerd_lock_fds;
-
/* communication with splice */
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.
+ */
+
/* timeout parameter, to control the polling thread grace period */
static int kconsumerd_poll_timeout = -1;
/* socket to exchange commands with sessiond */
static char *kconsumerd_command_sock_path;
-/* flag to inform the polling thread to kconsumerd_quit when all fd hung up */
-static int kconsumerd_quit = 0;
+/*
+ * flag to inform the polling thread to quit when all fd hung up.
+ * Updated by the kconsumerd_thread_receive_fds when it notices that all
+ * fds has hung up. Also updated by the signal handler
+ * (kconsumerd_should_exit()). Read by the polling threads.
+ */
+static volatile int kconsumerd_quit = 0;
/*
* kconsumerd_set_error_socket
kconsumerd_command_sock_path = sock;
}
+/*
+ * kconsumerd_find_session_fd
+ *
+ * Find a session fd in the global list.
+ *
+ * Return 1 if found else 0
+ */
+static int kconsumerd_find_session_fd(int fd)
+{
+ 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;
+}
+
/*
* kconsumerd_del_fd
*
*/
static void kconsumerd_del_fd(struct kconsumerd_fd *lcf)
{
- pthread_mutex_lock(&kconsumerd_lock_fds);
+ pthread_mutex_lock(&kconsumerd_data.lock);
cds_list_del(&lcf->list);
- if (kconsumerd_fds_count > 0) {
- kconsumerd_fds_count--;
+ if (kconsumerd_data.fds_count > 0) {
+ kconsumerd_data.fds_count--;
if (lcf != NULL) {
close(lcf->out_fd);
close(lcf->consumerd_fd);
lcf = NULL;
}
}
- pthread_mutex_unlock(&kconsumerd_lock_fds);
+ kconsumerd_data.need_update = 1;
+ pthread_mutex_unlock(&kconsumerd_data.lock);
}
/*
*/
static int kconsumerd_add_fd(struct lttcomm_kconsumerd_msg *buf, int consumerd_fd)
{
- struct kconsumerd_fd *tmp_fd;
int ret;
+ struct kconsumerd_fd *tmp_fd;
+
+ pthread_mutex_lock(&kconsumerd_data.lock);
+ /* Check if already exist */
+ ret = kconsumerd_find_session_fd(buf->fd);
+ if (ret == 1) {
+ goto end;
+ }
tmp_fd = malloc(sizeof(struct kconsumerd_fd));
tmp_fd->sessiond_fd = buf->fd;
DBG("Adding %s (%d, %d, %d)", tmp_fd->path_name,
tmp_fd->sessiond_fd, tmp_fd->consumerd_fd, tmp_fd->out_fd);
- pthread_mutex_lock(&kconsumerd_lock_fds);
- cds_list_add(&tmp_fd->list, &kconsumerd_fd_list.head);
- kconsumerd_fds_count++;
- pthread_mutex_unlock(&kconsumerd_lock_fds);
-
+ cds_list_add(&tmp_fd->list, &kconsumerd_data.fd_list.head);
+ kconsumerd_data.fds_count++;
+ kconsumerd_data.need_update = 1;
end:
+ pthread_mutex_unlock(&kconsumerd_data.lock);
return ret;
}
enum kconsumerd_fd_state state)
{
struct kconsumerd_fd *iter;
- cds_list_for_each_entry(iter, &kconsumerd_fd_list.head, list) {
+
+ pthread_mutex_lock(&kconsumerd_data.lock);
+ cds_list_for_each_entry(iter, &kconsumerd_data.fd_list.head, list) {
if (iter->sessiond_fd == sessiond_fd) {
iter->state = state;
break;
}
}
+ kconsumerd_data.need_update = 1;
+ pthread_mutex_unlock(&kconsumerd_data.lock);
}
/*
* to avoid doing a lookup in the linked list and concurrency issues
* when writing is needed.
* Returns the number of fds in the structures
+ * Called with kconsumerd_data.lock held.
*/
static int kconsumerd_update_poll_array(struct pollfd **pollfd,
struct kconsumerd_fd **local_kconsumerd_fd)
int i = 0;
DBG("Updating poll fd array");
- pthread_mutex_lock(&kconsumerd_lock_fds);
- cds_list_for_each_entry(iter, &kconsumerd_fd_list.head, list) {
+ cds_list_for_each_entry(iter, &kconsumerd_data.fd_list.head, list) {
DBG("Inside for each");
if (iter->state == ACTIVE_FD) {
DBG("Active FD %d", iter->consumerd_fd);
*/
(*pollfd)[i].fd = kconsumerd_poll_pipe[0];
(*pollfd)[i].events = POLLIN;
-
- kconsumerd_update_fd_array = 0;
- pthread_mutex_unlock(&kconsumerd_lock_fds);
return i;
}
default:
break;
}
- /* flag to tell the polling thread to update its fd array */
- kconsumerd_update_fd_array = 1;
/* signal the poll thread */
tmp2 = write(kconsumerd_poll_pipe[1], "4", 1);
} else {
}
end:
- DBG("kconsumerd_consumerd_recv_fd thread exiting");
return ret;
}
struct pollfd *pollfd = NULL;
/* local view of the fds */
struct kconsumerd_fd **local_kconsumerd_fd = NULL;
- /* local view of kconsumerd_fds_count */
+ /* local view of kconsumerd_data.fds_count */
int nb_fd = 0;
char tmp;
int tmp2;
* the ltt_fd_list has been updated, we need to update our
* local array as well
*/
- if (kconsumerd_update_fd_array == 1) {
+ pthread_mutex_lock(&kconsumerd_data.lock);
+ if (kconsumerd_data.need_update) {
if (pollfd != NULL) {
free(pollfd);
pollfd = NULL;
free(local_kconsumerd_fd);
local_kconsumerd_fd = NULL;
}
+
/* allocate for all fds + 1 for the kconsumerd_poll_pipe */
- pollfd = malloc((kconsumerd_fds_count + 1) * sizeof(struct pollfd));
+ pollfd = malloc((kconsumerd_data.fds_count + 1) * sizeof(struct pollfd));
if (pollfd == NULL) {
perror("pollfd malloc");
+ pthread_mutex_unlock(&kconsumerd_data.lock);
goto end;
}
+
/* allocate for all fds + 1 for the kconsumerd_poll_pipe */
- local_kconsumerd_fd = malloc((kconsumerd_fds_count + 1) *
+ local_kconsumerd_fd = malloc((kconsumerd_data.fds_count + 1) *
sizeof(struct kconsumerd_fd));
if (local_kconsumerd_fd == NULL) {
perror("local_kconsumerd_fd malloc");
+ pthread_mutex_unlock(&kconsumerd_data.lock);
goto end;
}
ret = kconsumerd_update_poll_array(&pollfd, local_kconsumerd_fd);
if (ret < 0) {
ERR("Error in allocating pollfd or local_outfds");
kconsumerd_send_error(KCONSUMERD_POLL_ERROR);
+ pthread_mutex_unlock(&kconsumerd_data.lock);
goto end;
}
nb_fd = ret;
+ kconsumerd_data.need_update = 0;
}
+ pthread_mutex_unlock(&kconsumerd_data.lock);
/* poll on the array of fds */
DBG("polling on %d fd", nb_fd + 1);
}
/*
- * if only the kconsumerd_poll_pipe triggered poll to return just
- * return to the beginning of the loop to update the array
+ * If the kconsumerd_poll_pipe triggered poll go
+ * directly to the beginning of the loop to update the
+ * array. We want to prioritize array update over
+ * low-priority reads.
*/
- if (num_rdy == 1 && pollfd[nb_fd].revents == POLLIN) {
+ if (pollfd[nb_fd].revents == POLLIN) {
DBG("kconsumerd_poll_pipe wake up");
tmp2 = read(kconsumerd_poll_pipe[0], &tmp, 1);
continue;
case POLLERR:
ERR("Error returned in polling fd %d.", pollfd[i].fd);
kconsumerd_del_fd(local_kconsumerd_fd[i]);
- kconsumerd_update_fd_array = 1;
num_hup++;
break;
case POLLHUP:
DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
kconsumerd_del_fd(local_kconsumerd_fd[i]);
- kconsumerd_update_fd_array = 1;
num_hup++;
break;
case POLLNVAL:
ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
kconsumerd_del_fd(local_kconsumerd_fd[i]);
- kconsumerd_update_fd_array = 1;
num_hup++;
break;
case POLLPRI:
free(local_kconsumerd_fd);
local_kconsumerd_fd = NULL;
}
- kconsumerd_cleanup();
return NULL;
}
goto end;
}
+ /* TODO: poll on socket and "should_quit" fd pipe */
+ /* TODO: change blocking call into non-blocking call */
/* Blocking call, waiting for transmission */
sock = lttcomm_accept_unix_sock(client_socket);
if (sock <= 0) {
}
while (1) {
/* 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("Received STOP command");
goto end;
}
+ if (kconsumerd_quit) {
+ 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);
if (ret <= 0) {
*
* Cleanup the daemon's socket on exit
*/
-void kconsumerd_cleanup()
+void kconsumerd_cleanup(void)
{
struct kconsumerd_fd *iter;
/* remove the socket file */
unlink(kconsumerd_command_sock_path);
- /* close all outfd */
- cds_list_for_each_entry(iter, &kconsumerd_fd_list.head, list) {
+ /*
+ * close all outfd. Called when there are no more threads
+ * 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) {
kconsumerd_del_fd(iter);
}
}
+/*
+ * Called from signal handler.
+ */
+void kconsumerd_should_exit(void)
+{
+ kconsumerd_quit = 1;
+ /*
+ * TODO: write into a should_quit pipe to wake up the fd
+ * receiver thread.
+ */
+}
+
/*
* kconsumerd_send_error
*
projects / lttng-tools.git / blobdiff