X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=68d69bb2a658658ef37dbd83b9c2094a40a4924a;hp=3455f827b6b715c5415bcb3ee11c0d6147a46972;hb=92b7a7f81a94f7616c155aed8a213edd68937a57;hpb=fa27abe8c16d7d47a901a8905e7cfd08c2e366d7 diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index 3455f827b..68d69bb2a 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -1082,78 +1082,46 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } break; } - case LTTNG_CONSUMER_SET_CHANNEL_ROTATE_PIPE: - { - int channel_rotate_pipe; - int flags; - - ret_code = LTTCOMM_CONSUMERD_SUCCESS; - /* Successfully received the command's type. */ - ret = consumer_send_status_msg(sock, ret_code); - if (ret < 0) { - goto error_fatal; - } - - ret = lttcomm_recv_fds_unix_sock(sock, &channel_rotate_pipe, 1); - if (ret != (ssize_t) sizeof(channel_rotate_pipe)) { - ERR("Failed to receive channel rotate pipe"); - goto error_fatal; - } - - DBG("Received channel rotate pipe (%d)", channel_rotate_pipe); - ctx->channel_rotate_pipe = channel_rotate_pipe; - /* Set the pipe as non-blocking. */ - ret = fcntl(channel_rotate_pipe, F_GETFL, 0); - if (ret == -1) { - PERROR("fcntl get flags of the channel rotate pipe"); - goto error_fatal; - } - flags = ret; - - ret = fcntl(channel_rotate_pipe, F_SETFL, flags | O_NONBLOCK); - if (ret == -1) { - PERROR("fcntl set O_NONBLOCK flag of the channel rotate pipe"); - goto error_fatal; - } - DBG("Channel rotate pipe set as non-blocking"); - ret_code = LTTCOMM_CONSUMERD_SUCCESS; - ret = consumer_send_status_msg(sock, ret_code); - if (ret < 0) { - goto error_fatal; - } - break; - } case LTTNG_CONSUMER_ROTATE_CHANNEL: { - DBG("Consumer rotate channel %" PRIu64, msg.u.rotate_channel.key); + struct lttng_consumer_channel *channel; + uint64_t key = msg.u.rotate_channel.key; - /* - * Sample the rotate position of all the streams in this channel. - */ - ret = lttng_consumer_rotate_channel(msg.u.rotate_channel.key, - msg.u.rotate_channel.pathname, - msg.u.rotate_channel.relayd_id, - msg.u.rotate_channel.metadata, - msg.u.rotate_channel.new_chunk_id, - ctx); - if (ret < 0) { - ERR("Rotate channel failed"); - ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; - } + DBG("Consumer rotate channel %" PRIu64, key); - health_code_update(); + channel = consumer_find_channel(key); + if (!channel) { + ERR("Channel %" PRIu64 " not found", key); + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + } else { + /* + * Sample the rotate position of all the streams in this channel. + */ + ret = lttng_consumer_rotate_channel(channel, key, + msg.u.rotate_channel.pathname, + msg.u.rotate_channel.relayd_id, + msg.u.rotate_channel.metadata, + msg.u.rotate_channel.new_chunk_id, + ctx); + if (ret < 0) { + ERR("Rotate channel failed"); + ret_code = LTTCOMM_CONSUMERD_ROTATION_FAIL; + } + health_code_update(); + } ret = consumer_send_status_msg(sock, ret_code); if (ret < 0) { /* Somehow, the session daemon is not responding anymore. */ goto end_nosignal; } - - /* Rotate the streams that are ready right now. */ - ret = lttng_consumer_rotate_ready_streams( - msg.u.rotate_channel.key, ctx); - if (ret < 0) { - ERR("Rotate ready streams failed"); + if (channel) { + /* Rotate the streams that are ready right now. */ + ret = lttng_consumer_rotate_ready_streams( + channel, key, ctx); + if (ret < 0) { + ERR("Rotate ready streams failed"); + } } break; @@ -1171,7 +1139,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, msg.u.rotate_rename.relayd_id); if (ret < 0) { ERR("Rotate rename failed"); - ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + ret_code = LTTCOMM_CONSUMERD_ROTATE_RENAME_FAILED; } health_code_update(); @@ -1183,20 +1151,19 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } break; } - case LTTNG_CONSUMER_ROTATE_PENDING_RELAY: + case LTTNG_CONSUMER_CHECK_ROTATION_PENDING_LOCAL: { int pending; uint32_t pending_reply; - DBG("Consumer rotate pending on relay for session %" PRIu64, - msg.u.rotate_pending_relay.session_id); - pending = lttng_consumer_rotate_pending_relay( - msg.u.rotate_pending_relay.session_id, - msg.u.rotate_pending_relay.relayd_id, - msg.u.rotate_pending_relay.chunk_id); + DBG("Perform local check of pending rotation for session id %" PRIu64, + msg.u.check_rotation_pending_local.session_id); + pending = lttng_consumer_check_rotation_pending_local( + msg.u.check_rotation_pending_local.session_id, + msg.u.check_rotation_pending_local.chunk_id); if (pending < 0) { - ERR("Rotate pending relay failed"); - ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + ERR("Local rotation pending check failed with code %i", pending); + ret_code = LTTCOMM_CONSUMERD_ROTATION_PENDING_LOCAL_FAILED; } else { pending_reply = !!pending; } @@ -1222,7 +1189,51 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ret = lttcomm_send_unix_sock(sock, &pending_reply, sizeof(pending_reply)); if (ret < 0) { - PERROR("send data pending ret code"); + PERROR("Failed to send rotation pending return code"); + goto error_fatal; + } + break; + } + case LTTNG_CONSUMER_CHECK_ROTATION_PENDING_RELAY: + { + int pending; + uint32_t pending_reply; + + DBG("Perform relayd check of pending rotation for session id %" PRIu64, + msg.u.check_rotation_pending_relay.session_id); + pending = lttng_consumer_check_rotation_pending_relay( + msg.u.check_rotation_pending_relay.session_id, + msg.u.check_rotation_pending_relay.relayd_id, + msg.u.check_rotation_pending_relay.chunk_id); + if (pending < 0) { + ERR("Relayd rotation pending check failed with code %i", pending); + ret_code = LTTCOMM_CONSUMERD_ROTATION_PENDING_RELAY_FAILED; + } else { + pending_reply = !!pending; + } + + health_code_update(); + + ret = consumer_send_status_msg(sock, ret_code); + if (ret < 0) { + /* Somehow, the session daemon is not responding anymore. */ + goto end_nosignal; + } + + if (pending < 0) { + /* + * An error occured while running the command; + * don't send the 'pending' flag as the sessiond + * will not read it. + */ + break; + } + + /* Send back returned value to session daemon */ + ret = lttcomm_send_unix_sock(sock, &pending_reply, + sizeof(pending_reply)); + if (ret < 0) { + PERROR("Failed to send rotation pending return code"); goto error_fatal; } break; @@ -1238,7 +1249,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, msg.u.mkdir.relayd_id); if (ret < 0) { ERR("consumer mkdir failed"); - ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + ret_code = LTTCOMM_CONSUMERD_MKDIR_FAILED; } health_code_update();