X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fconsumer.c;h=2e20878856fd1ee1ab1637b6ff297044bda58d77;hp=57b5b19d17db507b6b42d5407be7535512236e1e;hb=0b2dc8df2a6d7b3341a72a04767dd6328907c97c;hpb=7972aab22f74b18faa168c0482216a3dd711a075 diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c index 57b5b19d1..2e2087885 100644 --- a/src/bin/lttng-sessiond/consumer.c +++ b/src/bin/lttng-sessiond/consumer.c @@ -32,6 +32,7 @@ #include "consumer.h" #include "health.h" #include "ust-app.h" +#include "utils.h" /* * Receive a reply command status message from the consumer. Consumer socket @@ -407,6 +408,8 @@ error: /* * Delete the consumer_output object from the list and free the ptr. + * + * Should *NOT* be called with RCU read-side lock held. */ void consumer_destroy_output(struct consumer_output *obj) { @@ -426,7 +429,7 @@ void consumer_destroy_output(struct consumer_output *obj) rcu_read_unlock(); /* Finally destroy HT */ - lttng_ht_destroy(obj->socks); + ht_cleanup_push(obj->socks); } free(obj); @@ -434,6 +437,8 @@ void consumer_destroy_output(struct consumer_output *obj) /* * Copy consumer output and returned the newly allocated copy. + * + * Should *NOT* be called with RCU read-side lock held. */ struct consumer_output *consumer_copy_output(struct consumer_output *obj) { @@ -504,6 +509,12 @@ int consumer_set_network_uri(struct consumer_output *obj, if (uri->port == 0) { /* Assign default port. */ uri->port = DEFAULT_NETWORK_CONTROL_PORT; + } else { + if (obj->dst.net.data_isset && uri->port == + obj->dst.net.data.port) { + ret = -LTTNG_ERR_INVALID; + goto error; + } } DBG3("Consumer control URI set with port %d", uri->port); break; @@ -513,11 +524,18 @@ int consumer_set_network_uri(struct consumer_output *obj, if (uri->port == 0) { /* Assign default port. */ uri->port = DEFAULT_NETWORK_DATA_PORT; + } else { + if (obj->dst.net.control_isset && uri->port == + obj->dst.net.control.port) { + ret = -LTTNG_ERR_INVALID; + goto error; + } } DBG3("Consumer data URI set with port %d", uri->port); break; default: ERR("Set network uri type unknown %d", uri->stype); + ret = -LTTNG_ERR_INVALID; goto error; } @@ -553,6 +571,7 @@ int consumer_set_network_uri(struct consumer_output *obj, } if (ret < 0) { PERROR("snprintf set consumer uri subdir"); + ret = -LTTNG_ERR_NOMEM; goto error; } @@ -564,7 +583,7 @@ int consumer_set_network_uri(struct consumer_output *obj, equal: return 1; error: - return -1; + return ret; } /* @@ -663,7 +682,9 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, uint64_t relayd_id, uint64_t key, unsigned char *uuid, - uint32_t chan_id) + uint32_t chan_id, + uint64_t tracefile_size, + uint64_t tracefile_count) { assert(msg); @@ -684,6 +705,8 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, msg->u.ask_channel.relayd_id = relayd_id; msg->u.ask_channel.key = key; msg->u.ask_channel.chan_id = chan_id; + msg->u.ask_channel.tracefile_size = tracefile_size; + msg->u.ask_channel.tracefile_count = tracefile_count; memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid)); @@ -709,7 +732,9 @@ void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, const char *name, unsigned int nb_init_streams, enum lttng_event_output output, - int type) + int type, + uint64_t tracefile_size, + uint64_t tracefile_count) { assert(msg); @@ -726,6 +751,8 @@ void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, msg->u.channel.nb_init_streams = nb_init_streams; msg->u.channel.output = output; msg->u.channel.type = type; + msg->u.channel.tracefile_size = tracefile_size; + msg->u.channel.tracefile_count = tracefile_count; strncpy(msg->u.channel.pathname, pathname, sizeof(msg->u.channel.pathname)); @@ -797,14 +824,14 @@ error: * On success return positive value. On error, negative value. */ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock, - struct lttcomm_sock *sock, struct consumer_output *consumer, + struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer, enum lttng_stream_type type, uint64_t session_id) { int ret; struct lttcomm_consumer_msg msg; /* Code flow error. Safety net. */ - assert(sock); + assert(rsock); assert(consumer); assert(consumer_sock); @@ -823,13 +850,13 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock, msg.u.relayd_sock.net_index = consumer->net_seq_index; msg.u.relayd_sock.type = type; msg.u.relayd_sock.session_id = session_id; - memcpy(&msg.u.relayd_sock.sock, sock, sizeof(msg.u.relayd_sock.sock)); + 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)); if (ret < 0) { /* The above call will print a PERROR on error. */ - DBG("Error when sending relayd sockets on sock %d", sock->fd); + DBG("Error when sending relayd sockets on sock %d", rsock->sock.fd); goto error; } @@ -839,7 +866,7 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock, } DBG3("Sending relayd socket file descriptor to consumer"); - ret = consumer_send_fds(consumer_sock, &sock->fd, 1); + ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1); if (ret < 0) { goto error; } @@ -1072,7 +1099,7 @@ end: } /* - * Send metadata string to consumer. + * Send metadata string to consumer. Socket lock MUST be acquired. * * Return 0 on success else a negative value. */ @@ -1093,21 +1120,13 @@ int consumer_push_metadata(struct consumer_socket *socket, msg.u.push_metadata.target_offset = target_offset; msg.u.push_metadata.len = len; - /* - * TODO: reenable these locks when the consumerd gets the ability to - * reorder the metadata it receives. This fits with locking in - * src/bin/lttng-sessiond/ust-app.c:push_metadata() - * - * pthread_mutex_lock(socket->lock); - */ - health_code_update(); ret = consumer_send_msg(socket, &msg); - if (ret < 0) { + if (ret < 0 || len == 0) { goto end; } - DBG3("Consumer pushing metadata on sock %d of len %lu", 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); if (ret < 0) { @@ -1122,8 +1141,5 @@ int consumer_push_metadata(struct consumer_socket *socket, end: health_code_update(); - /* - * pthread_mutex_unlock(socket->lock); - */ return ret; }