From 4ce514c43483ba24fd935024da5b7aca681a7e52 Mon Sep 17 00:00:00 2001 From: David Goulet Date: Mon, 26 Aug 2013 12:26:10 -0400 Subject: [PATCH] Use consumer fd reference in consumer socket obj This is done so we don't copy the fd value of the consumer thus having a central control. This is part of a set of patches that makes the session daemon stop sending command to an invalid consumer (dead, closed). Acked-by: Mathieu Desnoyers Signed-off-by: David Goulet --- src/bin/lttng-sessiond/cmd.c | 10 +-- src/bin/lttng-sessiond/consumer.c | 79 +++++++++++++----------- src/bin/lttng-sessiond/consumer.h | 17 ++++- src/bin/lttng-sessiond/kernel-consumer.c | 4 +- src/bin/lttng-sessiond/kernel.c | 2 +- src/bin/lttng-sessiond/main.c | 21 +++---- src/bin/lttng-sessiond/ust-consumer.c | 18 +++--- 7 files changed, 83 insertions(+), 68 deletions(-) diff --git a/src/bin/lttng-sessiond/cmd.c b/src/bin/lttng-sessiond/cmd.c index 6a195e1bd..dcb5da5f4 100644 --- a/src/bin/lttng-sessiond/cmd.c +++ b/src/bin/lttng-sessiond/cmd.c @@ -463,7 +463,7 @@ static int init_kernel_tracing(struct ltt_kernel_session *session) cds_lfht_for_each_entry(session->consumer->socks->ht, &iter.iter, socket, node.node) { /* Code flow error */ - assert(socket->fd >= 0); + assert(socket->fd); pthread_mutex_lock(socket->lock); ret = kernel_consumer_send_session(socket, session); @@ -675,7 +675,7 @@ int cmd_setup_relayd(struct ltt_session *session) cds_lfht_for_each_entry(usess->consumer->socks->ht, &iter.iter, socket, node.node) { /* Code flow error */ - assert(socket->fd >= 0); + assert(socket->fd); pthread_mutex_lock(socket->lock); ret = send_consumer_relayd_sockets(LTTNG_DOMAIN_UST, session->id, @@ -694,7 +694,7 @@ int cmd_setup_relayd(struct ltt_session *session) cds_lfht_for_each_entry(ksess->consumer->socks->ht, &iter.iter, socket, node.node) { /* Code flow error */ - assert(socket->fd >= 0); + assert(socket->fd); pthread_mutex_lock(socket->lock); ret = send_consumer_relayd_sockets(LTTNG_DOMAIN_KERNEL, session->id, @@ -2027,13 +2027,15 @@ int cmd_register_consumer(struct ltt_session *session, int domain, ret = LTTNG_ERR_CONNECT_FAIL; goto error; } + cdata->cmd_sock = sock; - socket = consumer_allocate_socket(sock); + socket = consumer_allocate_socket(&cdata->cmd_sock); if (socket == NULL) { ret = close(sock); if (ret < 0) { PERROR("close register consumer"); } + cdata->cmd_sock = -1; ret = LTTNG_ERR_FATAL; goto error; } diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c index 17fc91976..39a194638 100644 --- a/src/bin/lttng-sessiond/consumer.c +++ b/src/bin/lttng-sessiond/consumer.c @@ -48,14 +48,14 @@ int consumer_recv_status_reply(struct consumer_socket *sock) assert(sock); - ret = lttcomm_recv_unix_sock(sock->fd, &reply, sizeof(reply)); + ret = lttcomm_recv_unix_sock(*sock->fd, &reply, sizeof(reply)); if (ret <= 0) { if (ret == 0) { /* Orderly shutdown. Don't return 0 which means success. */ ret = -1; } /* The above call will print a PERROR on error. */ - DBG("Fail to receive status reply on sock %d", sock->fd); + DBG("Fail to receive status reply on sock %d", *sock->fd); goto end; } @@ -89,14 +89,14 @@ int consumer_recv_status_channel(struct consumer_socket *sock, assert(stream_count); assert(key); - ret = lttcomm_recv_unix_sock(sock->fd, &reply, sizeof(reply)); + ret = lttcomm_recv_unix_sock(*sock->fd, &reply, sizeof(reply)); if (ret <= 0) { if (ret == 0) { /* Orderly shutdown. Don't return 0 which means success. */ ret = -1; } /* The above call will print a PERROR on error. */ - DBG("Fail to receive status reply on sock %d", sock->fd); + DBG("Fail to receive status reply on sock %d", *sock->fd); goto end; } @@ -127,7 +127,7 @@ int consumer_send_destroy_relayd(struct consumer_socket *sock, assert(consumer); assert(sock); - DBG2("Sending destroy relayd command to consumer sock %d", sock->fd); + DBG2("Sending destroy relayd command to consumer sock %d", *sock->fd); /* Bail out if consumer is disabled */ if (!consumer->enabled) { @@ -140,7 +140,7 @@ int consumer_send_destroy_relayd(struct consumer_socket *sock, msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index; pthread_mutex_lock(sock->lock); - ret = lttcomm_send_unix_sock(sock->fd, &msg, sizeof(msg)); + ret = lttcomm_send_unix_sock(*sock->fd, &msg, sizeof(msg)); if (ret < 0) { /* Indicate that the consumer is probably closing at this point. */ DBG("send consumer destroy relayd command"); @@ -213,7 +213,7 @@ int consumer_create_socket(struct consumer_data *data, socket = consumer_find_socket(data->cmd_sock, output); rcu_read_unlock(); if (socket == NULL) { - socket = consumer_allocate_socket(data->cmd_sock); + socket = consumer_allocate_socket(&data->cmd_sock); if (socket == NULL) { ret = -1; goto error; @@ -300,10 +300,12 @@ struct consumer_socket *consumer_find_socket(int key, /* * Allocate a new consumer_socket and return the pointer. */ -struct consumer_socket *consumer_allocate_socket(int fd) +struct consumer_socket *consumer_allocate_socket(int *fd) { struct consumer_socket *socket = NULL; + assert(fd); + socket = zmalloc(sizeof(struct consumer_socket)); if (socket == NULL) { PERROR("zmalloc consumer socket"); @@ -311,7 +313,7 @@ struct consumer_socket *consumer_allocate_socket(int fd) } socket->fd = fd; - lttng_ht_node_init_ulong(&socket->node, fd); + lttng_ht_node_init_ulong(&socket->node, *fd); error: return socket; @@ -375,8 +377,8 @@ void consumer_destroy_socket(struct consumer_socket *sock) * consumer was registered, */ if (sock->registered) { - DBG3("Consumer socket was registered. Closing fd %d", sock->fd); - lttcomm_close_unix_sock(sock->fd); + DBG3("Consumer socket was registered. Closing fd %d", *sock->fd); + lttcomm_close_unix_sock(*sock->fd); } call_rcu(&sock->node.head, destroy_socket_rcu); @@ -507,7 +509,7 @@ int consumer_copy_sockets(struct consumer_output *dst, rcu_read_lock(); cds_lfht_for_each_entry(src->socks->ht, &iter.iter, socket, node.node) { /* Ignore socket that are already there. */ - copy_sock = consumer_find_socket(socket->fd, dst); + copy_sock = consumer_find_socket(*socket->fd, dst); if (copy_sock) { continue; } @@ -646,12 +648,13 @@ int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd) assert(fds); assert(sock); + assert(sock->fd); assert(nb_fd > 0); - ret = lttcomm_send_fds_unix_sock(sock->fd, fds, nb_fd); + ret = lttcomm_send_fds_unix_sock(*sock->fd, fds, nb_fd); if (ret < 0) { /* The above call will print a PERROR on error. */ - DBG("Error when sending consumer fds on sock %d", sock->fd); + DBG("Error when sending consumer fds on sock %d", *sock->fd); goto error; } @@ -671,13 +674,13 @@ int consumer_send_msg(struct consumer_socket *sock, assert(msg); assert(sock); - assert(sock->fd >= 0); + assert(sock->fd); - ret = lttcomm_send_unix_sock(sock->fd, msg, + ret = lttcomm_send_unix_sock(*sock->fd, msg, sizeof(struct lttcomm_consumer_msg)); if (ret < 0) { /* The above call will print a PERROR on error. */ - DBG("Error when sending consumer channel on sock %d", sock->fd); + DBG("Error when sending consumer channel on sock %d", *sock->fd); goto error; } @@ -697,13 +700,13 @@ int consumer_send_channel(struct consumer_socket *sock, assert(msg); assert(sock); - assert(sock->fd >= 0); + assert(sock->fd); - ret = lttcomm_send_unix_sock(sock->fd, msg, + ret = lttcomm_send_unix_sock(*sock->fd, msg, sizeof(struct lttcomm_consumer_msg)); if (ret < 0) { /* The above call will print a PERROR on error. */ - DBG("Error when sending consumer channel on sock %d", sock->fd); + DBG("Error when sending consumer channel on sock %d", *sock->fd); goto error; } @@ -854,14 +857,15 @@ int consumer_send_stream(struct consumer_socket *sock, assert(msg); assert(dst); assert(sock); + assert(sock->fd); assert(fds); /* Send on socket */ - ret = lttcomm_send_unix_sock(sock->fd, msg, + ret = lttcomm_send_unix_sock(*sock->fd, msg, sizeof(struct lttcomm_consumer_msg)); if (ret < 0) { /* The above call will print a PERROR on error. */ - DBG("Error when sending consumer stream on sock %d", sock->fd); + DBG("Error when sending consumer stream on sock %d", *sock->fd); goto error; } @@ -895,6 +899,7 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock, assert(rsock); assert(consumer); assert(consumer_sock); + assert(consumer_sock->fd); /* Bail out if consumer is disabled */ if (!consumer->enabled) { @@ -913,8 +918,8 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock, msg.u.relayd_sock.session_id = session_id; memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock)); - DBG3("Sending relayd sock info to consumer on %d", consumer_sock->fd); - ret = lttcomm_send_unix_sock(consumer_sock->fd, &msg, sizeof(msg)); + DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd); + ret = lttcomm_send_unix_sock(*consumer_sock->fd, &msg, sizeof(msg)); if (ret < 0) { /* The above call will print a PERROR on error. */ DBG("Error when sending relayd sockets on sock %d", rsock->sock.fd); @@ -1015,14 +1020,14 @@ int consumer_is_data_pending(uint64_t session_id, cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket, node.node) { /* Code flow error */ - assert(socket->fd >= 0); + assert(socket->fd); pthread_mutex_lock(socket->lock); - ret = lttcomm_send_unix_sock(socket->fd, &msg, sizeof(msg)); + ret = lttcomm_send_unix_sock(*socket->fd, &msg, sizeof(msg)); if (ret < 0) { /* The above call will print a PERROR on error. */ - DBG("Error on consumer is data pending on sock %d", socket->fd); + DBG("Error on consumer is data pending on sock %d", *socket->fd); pthread_mutex_unlock(socket->lock); goto error_unlock; } @@ -1032,14 +1037,14 @@ int consumer_is_data_pending(uint64_t session_id, * the reply status message. */ - ret = lttcomm_recv_unix_sock(socket->fd, &ret_code, sizeof(ret_code)); + ret = lttcomm_recv_unix_sock(*socket->fd, &ret_code, sizeof(ret_code)); if (ret <= 0) { if (ret == 0) { /* Orderly shutdown. Don't return 0 which means success. */ ret = -1; } /* The above call will print a PERROR on error. */ - DBG("Error on recv consumer is data pending on sock %d", socket->fd); + DBG("Error on recv consumer is data pending on sock %d", *socket->fd); pthread_mutex_unlock(socket->lock); goto error_unlock; } @@ -1072,7 +1077,7 @@ int consumer_flush_channel(struct consumer_socket *socket, uint64_t key) struct lttcomm_consumer_msg msg; assert(socket); - assert(socket->fd >= 0); + assert(socket->fd); DBG2("Consumer flush channel key %" PRIu64, key); @@ -1105,7 +1110,7 @@ int consumer_close_metadata(struct consumer_socket *socket, struct lttcomm_consumer_msg msg; assert(socket); - assert(socket->fd >= 0); + assert(socket->fd); DBG2("Consumer close metadata channel key %" PRIu64, metadata_key); @@ -1138,7 +1143,7 @@ int consumer_setup_metadata(struct consumer_socket *socket, struct lttcomm_consumer_msg msg; assert(socket); - assert(socket->fd >= 0); + assert(socket->fd); DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key); @@ -1172,9 +1177,9 @@ int consumer_push_metadata(struct consumer_socket *socket, struct lttcomm_consumer_msg msg; assert(socket); - assert(socket->fd >= 0); + assert(socket->fd); - DBG2("Consumer push metadata to consumer socket %d", socket->fd); + DBG2("Consumer push metadata to consumer socket %d", *socket->fd); msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA; msg.u.push_metadata.key = metadata_key; @@ -1187,9 +1192,9 @@ int consumer_push_metadata(struct consumer_socket *socket, goto end; } - DBG3("Consumer pushing metadata on sock %d of len %zu", socket->fd, len); + DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd, len); - ret = lttcomm_send_unix_sock(socket->fd, metadata_str, len); + ret = lttcomm_send_unix_sock(*socket->fd, metadata_str, len); if (ret < 0) { goto end; } @@ -1218,7 +1223,7 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key, struct lttcomm_consumer_msg msg; assert(socket); - assert(socket->fd >= 0); + assert(socket->fd); assert(output); assert(output->consumer); diff --git a/src/bin/lttng-sessiond/consumer.h b/src/bin/lttng-sessiond/consumer.h index 2dd3b81dd..7ff3e5dcf 100644 --- a/src/bin/lttng-sessiond/consumer.h +++ b/src/bin/lttng-sessiond/consumer.h @@ -33,8 +33,12 @@ enum consumer_dst_type { }; struct consumer_socket { - /* File descriptor */ - int fd; + /* + * File descriptor. This is just a reference to the consumer data meaning + * that every access must be locked and check for a possible invalid value. + */ + int *fd; + /* * To use this socket (send/recv), this lock MUST be acquired. */ @@ -87,6 +91,13 @@ struct consumer_data { int err_sock; /* These two sockets uses the cmd_unix_sock_path. */ int cmd_sock; + /* + * The metadata socket object is handled differently and only created + * locally in this object thus it's the only reference available in the + * session daemon. For that reason, a static variable for the fd is + * required and the metadata socket fd points to it. + */ + int metadata_fd; struct consumer_socket metadata_sock; /* consumer error and command Unix socket path */ @@ -157,7 +168,7 @@ struct consumer_socket *consumer_find_socket(int key, struct consumer_output *consumer); struct consumer_socket *consumer_find_socket_by_bitness(int bits, struct consumer_output *consumer); -struct consumer_socket *consumer_allocate_socket(int fd); +struct consumer_socket *consumer_allocate_socket(int *fd); void consumer_add_socket(struct consumer_socket *sock, struct consumer_output *consumer); void consumer_del_socket(struct consumer_socket *sock, diff --git a/src/bin/lttng-sessiond/kernel-consumer.c b/src/bin/lttng-sessiond/kernel-consumer.c index bf3559a21..1405554d9 100644 --- a/src/bin/lttng-sessiond/kernel-consumer.c +++ b/src/bin/lttng-sessiond/kernel-consumer.c @@ -371,7 +371,7 @@ int kernel_consumer_destroy_channel(struct consumer_socket *socket, assert(channel); assert(socket); - assert(socket->fd >= 0); + assert(socket->fd); DBG("Sending kernel consumer destroy channel key %d", channel->fd); @@ -400,7 +400,7 @@ int kernel_consumer_destroy_metadata(struct consumer_socket *socket, assert(metadata); assert(socket); - assert(socket->fd >= 0); + assert(socket->fd); DBG("Sending kernel consumer destroy channel key %d", metadata->fd); diff --git a/src/bin/lttng-sessiond/kernel.c b/src/bin/lttng-sessiond/kernel.c index bf1a52171..5d4f61451 100644 --- a/src/bin/lttng-sessiond/kernel.c +++ b/src/bin/lttng-sessiond/kernel.c @@ -864,7 +864,7 @@ int kernel_snapshot_record(struct ltt_kernel_session *ksess, struct consumer_output *saved_output; struct ltt_kernel_channel *chan; /* Code flow error */ - assert(socket->fd >= 0); + assert(socket->fd); /* * Temporarly switch consumer output for our snapshot output. As long diff --git a/src/bin/lttng-sessiond/main.c b/src/bin/lttng-sessiond/main.c index 4ebd1e7d6..b688a4c09 100644 --- a/src/bin/lttng-sessiond/main.c +++ b/src/bin/lttng-sessiond/main.c @@ -90,7 +90,6 @@ static struct consumer_data kconsumer_data = { .cmd_unix_sock_path = DEFAULT_KCONSUMERD_CMD_SOCK_PATH, .err_sock = -1, .cmd_sock = -1, - .metadata_sock.fd = -1, .pid_mutex = PTHREAD_MUTEX_INITIALIZER, .lock = PTHREAD_MUTEX_INITIALIZER, .cond = PTHREAD_COND_INITIALIZER, @@ -102,7 +101,6 @@ static struct consumer_data ustconsumer64_data = { .cmd_unix_sock_path = DEFAULT_USTCONSUMERD64_CMD_SOCK_PATH, .err_sock = -1, .cmd_sock = -1, - .metadata_sock.fd = -1, .pid_mutex = PTHREAD_MUTEX_INITIALIZER, .lock = PTHREAD_MUTEX_INITIALIZER, .cond = PTHREAD_COND_INITIALIZER, @@ -114,7 +112,6 @@ static struct consumer_data ustconsumer32_data = { .cmd_unix_sock_path = DEFAULT_USTCONSUMERD32_CMD_SOCK_PATH, .err_sock = -1, .cmd_sock = -1, - .metadata_sock.fd = -1, .pid_mutex = PTHREAD_MUTEX_INITIALIZER, .lock = PTHREAD_MUTEX_INITIALIZER, .cond = PTHREAD_COND_INITIALIZER, @@ -694,7 +691,7 @@ static int update_kernel_stream(struct consumer_data *consumer_data, int fd) cds_lfht_for_each_entry(ksess->consumer->socks->ht, &iter.iter, socket, node.node) { /* Code flow error */ - assert(socket->fd >= 0); + assert(socket->fd); pthread_mutex_lock(socket->lock); ret = kernel_consumer_send_channel_stream(socket, @@ -1019,15 +1016,15 @@ restart: /* Connect both socket, command and metadata. */ consumer_data->cmd_sock = lttcomm_connect_unix_sock(consumer_data->cmd_unix_sock_path); - consumer_data->metadata_sock.fd = + consumer_data->metadata_fd = lttcomm_connect_unix_sock(consumer_data->cmd_unix_sock_path); - if (consumer_data->cmd_sock < 0 || - consumer_data->metadata_sock.fd < 0) { + if (consumer_data->cmd_sock < 0 || consumer_data->metadata_fd < 0) { PERROR("consumer connect cmd socket"); /* On error, signal condition and quit. */ signal_consumer_condition(consumer_data, -1); goto error; } + consumer_data->metadata_sock.fd = &consumer_data->metadata_fd; /* Create metadata socket lock. */ consumer_data->metadata_sock.lock = zmalloc(sizeof(pthread_mutex_t)); if (consumer_data->metadata_sock.lock == NULL) { @@ -1040,7 +1037,7 @@ restart: signal_consumer_condition(consumer_data, 1); DBG("Consumer command socket ready (fd: %d", consumer_data->cmd_sock); DBG("Consumer metadata socket ready (fd: %d)", - consumer_data->metadata_sock.fd); + consumer_data->metadata_fd); } else { ERR("consumer error when waiting for SOCK_READY : %s", lttcomm_get_readable_code(-code)); @@ -1060,7 +1057,7 @@ restart: } /* Add metadata socket that is successfully connected. */ - ret = lttng_poll_add(&events, consumer_data->metadata_sock.fd, + ret = lttng_poll_add(&events, consumer_data->metadata_fd, LPOLLIN | LPOLLRDHUP); if (ret < 0) { goto error; @@ -1119,7 +1116,7 @@ restart_poll: lttcomm_get_readable_code(-code)); goto exit; - } else if (pollfd == consumer_data->metadata_sock.fd) { + } else if (pollfd == consumer_data->metadata_fd) { /* UST metadata requests */ ret = ust_consumer_metadata_request( &consumer_data->metadata_sock); @@ -1163,8 +1160,8 @@ error: } consumer_data->cmd_sock = -1; } - if (consumer_data->metadata_sock.fd >= 0) { - ret = close(consumer_data->metadata_sock.fd); + if (*consumer_data->metadata_sock.fd >= 0) { + ret = close(*consumer_data->metadata_sock.fd); if (ret) { PERROR("close"); } diff --git a/src/bin/lttng-sessiond/ust-consumer.c b/src/bin/lttng-sessiond/ust-consumer.c index d6fe58cd9..2eb8cda25 100644 --- a/src/bin/lttng-sessiond/ust-consumer.c +++ b/src/bin/lttng-sessiond/ust-consumer.c @@ -166,7 +166,7 @@ static int ask_channel_creation(struct ust_app_session *ua_sess, health_code_update(); - ret = lttcomm_send_unix_sock(socket->fd, &msg, sizeof(msg)); + ret = lttcomm_send_unix_sock(*socket->fd, &msg, sizeof(msg)); if (ret < 0) { goto error; } @@ -207,7 +207,7 @@ int ust_consumer_ask_channel(struct ust_app_session *ua_sess, assert(ua_chan); assert(consumer); assert(socket); - assert(socket->fd >= 0); + assert(socket->fd); assert(registry); if (!consumer->enabled) { @@ -242,7 +242,7 @@ int ust_consumer_get_channel(struct consumer_socket *socket, assert(ua_chan); assert(socket); - assert(socket->fd >= 0); + assert(socket->fd); msg.cmd_type = LTTNG_CONSUMER_GET_CHANNEL; msg.u.get_channel.key = ua_chan->key; @@ -257,11 +257,11 @@ int ust_consumer_get_channel(struct consumer_socket *socket, } /* First, get the channel from consumer. */ - ret = ustctl_recv_channel_from_consumer(socket->fd, &ua_chan->obj); + ret = ustctl_recv_channel_from_consumer(*socket->fd, &ua_chan->obj); if (ret < 0) { if (ret != -EPIPE) { ERR("Error recv channel from consumer %d with ret %d", - socket->fd, ret); + *socket->fd, ret); } else { DBG3("UST app recv channel from consumer. Consumer is dead."); } @@ -280,7 +280,7 @@ int ust_consumer_get_channel(struct consumer_socket *socket, } /* Stream object is populated by this call if successful. */ - ret = ustctl_recv_stream_from_consumer(socket->fd, &stream->obj); + ret = ustctl_recv_stream_from_consumer(*socket->fd, &stream->obj); if (ret < 0) { free(stream); if (ret == -LTTNG_UST_ERR_NOENT) { @@ -290,7 +290,7 @@ int ust_consumer_get_channel(struct consumer_socket *socket, } if (ret != -EPIPE) { ERR("Recv stream from consumer %d with ret %d", - socket->fd, ret); + *socket->fd, ret); } else { DBG3("UST app recv stream from consumer. Consumer is dead."); } @@ -337,7 +337,7 @@ int ust_consumer_destroy_channel(struct consumer_socket *socket, assert(ua_chan); assert(socket); - assert(socket->fd >= 0); + assert(socket->fd); msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL; msg.u.destroy_channel.key = ua_chan->key; @@ -445,7 +445,7 @@ int ust_consumer_metadata_request(struct consumer_socket *socket) health_code_update(); /* Wait for a metadata request */ - ret = lttcomm_recv_unix_sock(socket->fd, &request, sizeof(request)); + ret = lttcomm_recv_unix_sock(*socket->fd, &request, sizeof(request)); if (ret <= 0) { ERR("Consumer closed the metadata socket"); ret = -1; -- 2.34.1