X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fconsumer.c;h=f89bb1df290fbc8d5f6682e1f1089448974ca40e;hp=8727ce3776a1d85d80a9bdef7ae09110cedb256e;hb=d88744a44aa5f2ca90ab87946692b9eed3120641;hpb=00fb02ace5151a6546f4e97e5439512913a50e68 diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c index 8727ce377..f89bb1df2 100644 --- a/src/bin/lttng-sessiond/consumer.c +++ b/src/bin/lttng-sessiond/consumer.c @@ -1064,37 +1064,70 @@ error: return ret; } -int consumer_send_channel_monitor_pipe(struct consumer_socket *consumer_sock, - int pipe) +static +int consumer_send_pipe(struct consumer_socket *consumer_sock, + enum lttng_consumer_command cmd, int pipe) { int ret; struct lttcomm_consumer_msg msg; + const char *pipe_name; + const char *command_name; + + switch (cmd) { + case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE: + pipe_name = "channel monitor"; + command_name = "SET_CHANNEL_MONITOR_PIPE"; + break; + case LTTNG_CONSUMER_SET_CHANNEL_ROTATE_PIPE: + pipe_name = "channel rotate"; + command_name = "SET_CHANNEL_ROTATE_PIPE"; + break; + default: + ERR("Unexpected command received in %s (cmd = %d)", __func__, + (int) cmd); + abort(); + } /* Code flow error. Safety net. */ memset(&msg, 0, sizeof(msg)); - msg.cmd_type = LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE; + msg.cmd_type = cmd; pthread_mutex_lock(consumer_sock->lock); - DBG3("Sending set_channel_monitor_pipe command to consumer"); + DBG3("Sending %s command to consumer", command_name); ret = consumer_send_msg(consumer_sock, &msg); if (ret < 0) { goto error; } - DBG3("Sending channel monitoring pipe %d to consumer on socket %d", + DBG3("Sending %s pipe %d to consumer on socket %d", + pipe_name, pipe, *consumer_sock->fd_ptr); ret = consumer_send_fds(consumer_sock, &pipe, 1); if (ret < 0) { goto error; } - DBG2("Channel monitoring pipe successfully sent"); + DBG2("%s pipe successfully sent", pipe_name); error: pthread_mutex_unlock(consumer_sock->lock); return ret; } +int consumer_send_channel_monitor_pipe(struct consumer_socket *consumer_sock, + int pipe) +{ + return consumer_send_pipe(consumer_sock, + LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE, pipe); +} + +int consumer_send_channel_rotate_pipe(struct consumer_socket *consumer_sock, + int pipe) +{ + return consumer_send_pipe(consumer_sock, + LTTNG_CONSUMER_SET_CHANNEL_ROTATE_PIPE, pipe); +} + /* * Set consumer subdirectory using the session name and a generated datetime if * needed. This is appended to the current subdirectory. @@ -1580,6 +1613,72 @@ end: return ret; } +/* + * Ask the consumer to rotate a channel. + * domain_path contains "/kernel" for kernel or the complete path for UST + * (ex: /ust/uid/1000/64-bit); + * + * The new_chunk_id is the session->rotate_count that has been incremented + * when the rotation started. On the relay, this allows to keep track in which + * chunk each stream is currently writing to (for the rotate_pending operation). + */ +int consumer_rotate_channel(struct consumer_socket *socket, uint64_t key, + uid_t uid, gid_t gid, struct consumer_output *output, + char *domain_path, bool is_metadata_channel, + uint64_t new_chunk_id, + bool *rotate_pending_relay) +{ + int ret; + struct lttcomm_consumer_msg msg; + + assert(socket); + + DBG("Consumer rotate channel key %" PRIu64, key); + + pthread_mutex_lock(socket->lock); + memset(&msg, 0, sizeof(msg)); + msg.cmd_type = LTTNG_CONSUMER_ROTATE_CHANNEL; + msg.u.rotate_channel.key = key; + msg.u.rotate_channel.metadata = !!is_metadata_channel; + msg.u.rotate_channel.new_chunk_id = new_chunk_id; + + if (output->type == CONSUMER_DST_NET) { + msg.u.rotate_channel.relayd_id = output->net_seq_index; + ret = snprintf(msg.u.rotate_channel.pathname, + sizeof(msg.u.rotate_channel.pathname), "%s%s%s", + output->dst.net.base_dir, + output->chunk_path, domain_path); + if (ret < 0 || ret == sizeof(msg.u.rotate_channel.pathname)) { + ERR("Failed to format channel path name when asking consumer to rotate channel"); + ret = -1; + goto error; + } + *rotate_pending_relay = true; + } else { + msg.u.rotate_channel.relayd_id = (uint64_t) -1ULL; + ret = snprintf(msg.u.rotate_channel.pathname, + sizeof(msg.u.rotate_channel.pathname), "%s%s%s", + output->dst.session_root_path, + output->chunk_path, domain_path); + if (ret < 0 || ret == sizeof(msg.u.rotate_channel.pathname)) { + ERR("Failed to format channel path name when asking consumer to rotate channel"); + ret = -1; + goto error; + } + } + + health_code_update(); + ret = consumer_send_msg(socket, &msg); + if (ret < 0) { + goto error; + } + +error: + pthread_mutex_unlock(socket->lock); + health_code_update(); + return ret; +} + int consumer_rotate_rename(struct consumer_socket *socket, uint64_t session_id, const struct consumer_output *output, const char *old_path, const char *new_path, uid_t uid, gid_t gid) @@ -1636,6 +1735,51 @@ error: return ret; } +/* + * Ask the relay if a rotation is still pending. Must be called with the socket + * lock held. + * + * Return 1 if the rotation is still pending, 0 if finished, a negative value + * on error. + */ +int consumer_rotate_pending_relay(struct consumer_socket *socket, + struct consumer_output *output, uint64_t session_id, + uint64_t chunk_id) +{ + int ret; + struct lttcomm_consumer_msg msg; + uint32_t pending = 0; + + assert(socket); + + DBG("Consumer rotate pending on relay for session %" PRIu64 ", chunk id % " PRIu64, + session_id, chunk_id); + assert(output->type == CONSUMER_DST_NET); + + memset(&msg, 0, sizeof(msg)); + msg.cmd_type = LTTNG_CONSUMER_ROTATE_PENDING_RELAY; + msg.u.rotate_pending_relay.session_id = session_id; + msg.u.rotate_pending_relay.relayd_id = output->net_seq_index; + msg.u.rotate_pending_relay.chunk_id = chunk_id; + + health_code_update(); + ret = consumer_send_msg(socket, &msg); + if (ret < 0) { + goto error; + } + + ret = consumer_socket_recv(socket, &pending, sizeof(pending)); + if (ret < 0) { + goto error; + } + + ret = pending; + +error: + health_code_update(); + return ret; +} + /* * Ask the consumer to create a directory. *