X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fconsumer.c;h=2e20878856fd1ee1ab1637b6ff297044bda58d77;hp=ff5360aae755aa082ec6517cb6c4f10df99c9d9d;hb=0b2dc8df2a6d7b3341a72a04767dd6328907c97c;hpb=ffe600149a7608221985751e1bf293234bf2545c diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c index ff5360aae..2e2087885 100644 --- a/src/bin/lttng-sessiond/consumer.c +++ b/src/bin/lttng-sessiond/consumer.c @@ -23,12 +23,16 @@ #include #include #include +#include #include #include #include #include "consumer.h" +#include "health.h" +#include "ust-app.h" +#include "utils.h" /* * Receive a reply command status message from the consumer. Consumer socket @@ -76,7 +80,7 @@ end: * negative value is sent back and both parameters are untouched. */ int consumer_recv_status_channel(struct consumer_socket *sock, - unsigned long *key, unsigned int *stream_count) + uint64_t *key, unsigned int *stream_count) { int ret; struct lttcomm_consumer_status_channel reply; @@ -229,6 +233,41 @@ error: return ret; } +/* + * Return the consumer socket from the given consumer output with the right + * bitness. On error, returns NULL. + * + * The caller MUST acquire a rcu read side lock and keep it until the socket + * object reference is not needed anymore. + */ +struct consumer_socket *consumer_find_socket_by_bitness(int bits, + struct consumer_output *consumer) +{ + int consumer_fd; + struct consumer_socket *socket = NULL; + + switch (bits) { + case 64: + consumer_fd = uatomic_read(&ust_consumerd64_fd); + break; + case 32: + consumer_fd = uatomic_read(&ust_consumerd32_fd); + break; + default: + assert(0); + goto end; + } + + socket = consumer_find_socket(consumer_fd, consumer); + if (!socket) { + ERR("Consumer socket fd %d not found in consumer obj %p", + consumer_fd, consumer); + } + +end: + return socket; +} + /* * Find a consumer_socket in a consumer_output hashtable. Read side lock must * be acquired before calling this function and across use of the @@ -359,7 +398,7 @@ struct consumer_output *consumer_create_output(enum consumer_dst_type type) /* By default, consumer output is enabled */ output->enabled = 1; output->type = type; - output->net_seq_index = -1; + output->net_seq_index = (uint64_t) -1ULL; output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG); @@ -369,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) { @@ -388,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); @@ -396,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) { @@ -466,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; @@ -475,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; } @@ -515,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; } @@ -526,7 +583,7 @@ int consumer_set_network_uri(struct consumer_output *obj, equal: return 1; error: - return -1; + return ret; } /* @@ -622,9 +679,12 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, const char *name, uid_t uid, gid_t gid, - int relayd_id, - unsigned long key, - unsigned char *uuid) + uint64_t relayd_id, + uint64_t key, + unsigned char *uuid, + uint32_t chan_id, + uint64_t tracefile_size, + uint64_t tracefile_count) { assert(msg); @@ -644,6 +704,9 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, msg->u.ask_channel.gid = gid; 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)); @@ -660,16 +723,18 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, */ void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, enum lttng_consumer_command cmd, - int channel_key, + uint64_t channel_key, uint64_t session_id, const char *pathname, uid_t uid, gid_t gid, - int relayd_id, + uint64_t relayd_id, 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); @@ -686,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)); @@ -700,8 +767,8 @@ void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, */ void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg, enum lttng_consumer_command cmd, - int channel_key, - int stream_key, + uint64_t channel_key, + uint64_t stream_key, int cpu) { assert(msg); @@ -757,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, - enum lttng_stream_type type, unsigned int session_id) + 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); @@ -783,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; } @@ -799,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; } @@ -865,7 +932,7 @@ error: * This function has a different behavior with the consumer i.e. that it waits * for a reply from the consumer if yes or no the data is pending. */ -int consumer_is_data_pending(unsigned int id, +int consumer_is_data_pending(uint64_t session_id, struct consumer_output *consumer) { int ret; @@ -878,9 +945,9 @@ int consumer_is_data_pending(unsigned int id, msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING; - msg.u.data_pending.session_id = (uint64_t) id; + msg.u.data_pending.session_id = session_id; - DBG3("Consumer data pending for id %u", id); + DBG3("Consumer data pending for id %" PRIu64, session_id); /* Send command for each consumer */ rcu_read_lock(); @@ -924,11 +991,155 @@ int consumer_is_data_pending(unsigned int id, } rcu_read_unlock(); - DBG("Consumer data is %s pending for session id %u", - ret_code == 1 ? "" : "NOT", id); + DBG("Consumer data is %s pending for session id %" PRIu64, + ret_code == 1 ? "" : "NOT", session_id); return ret_code; error_unlock: rcu_read_unlock(); return -1; } + +/* + * Send a flush command to consumer using the given channel key. + * + * Return 0 on success else a negative value. + */ +int consumer_flush_channel(struct consumer_socket *socket, uint64_t key) +{ + int ret; + struct lttcomm_consumer_msg msg; + + assert(socket); + assert(socket->fd >= 0); + + DBG2("Consumer flush channel key %" PRIu64, key); + + msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL; + msg.u.flush_channel.key = key; + + pthread_mutex_lock(socket->lock); + health_code_update(); + + ret = consumer_send_msg(socket, &msg); + if (ret < 0) { + goto end; + } + +end: + health_code_update(); + pthread_mutex_unlock(socket->lock); + return ret; +} + +/* + * Send a close metdata command to consumer using the given channel key. + * + * Return 0 on success else a negative value. + */ +int consumer_close_metadata(struct consumer_socket *socket, + uint64_t metadata_key) +{ + int ret; + struct lttcomm_consumer_msg msg; + + assert(socket); + assert(socket->fd >= 0); + + DBG2("Consumer close metadata channel key %" PRIu64, metadata_key); + + msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA; + msg.u.close_metadata.key = metadata_key; + + pthread_mutex_lock(socket->lock); + health_code_update(); + + ret = consumer_send_msg(socket, &msg); + if (ret < 0) { + goto end; + } + +end: + health_code_update(); + pthread_mutex_unlock(socket->lock); + return ret; +} + +/* + * Send a setup metdata command to consumer using the given channel key. + * + * Return 0 on success else a negative value. + */ +int consumer_setup_metadata(struct consumer_socket *socket, + uint64_t metadata_key) +{ + int ret; + struct lttcomm_consumer_msg msg; + + assert(socket); + assert(socket->fd >= 0); + + DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key); + + msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA; + msg.u.setup_metadata.key = metadata_key; + + pthread_mutex_lock(socket->lock); + health_code_update(); + + ret = consumer_send_msg(socket, &msg); + if (ret < 0) { + goto end; + } + +end: + health_code_update(); + pthread_mutex_unlock(socket->lock); + return ret; +} + +/* + * Send metadata string to consumer. Socket lock MUST be acquired. + * + * Return 0 on success else a negative value. + */ +int consumer_push_metadata(struct consumer_socket *socket, + uint64_t metadata_key, char *metadata_str, size_t len, + size_t target_offset) +{ + int ret; + struct lttcomm_consumer_msg msg; + + assert(socket); + assert(socket->fd >= 0); + + DBG2("Consumer push metadata to consumer socket %d", socket->fd); + + msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA; + msg.u.push_metadata.key = metadata_key; + msg.u.push_metadata.target_offset = target_offset; + msg.u.push_metadata.len = len; + + health_code_update(); + ret = consumer_send_msg(socket, &msg); + if (ret < 0 || len == 0) { + goto end; + } + + 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) { + goto end; + } + + health_code_update(); + ret = consumer_recv_status_reply(socket); + if (ret < 0) { + goto end; + } + +end: + health_code_update(); + return ret; +}