X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=c2ad0fdd8cef613f170f4c43673d5d356fde8f93;hp=76238a087976ff5ca341040ecbafecfcb3d0b0a0;hb=c8f59ee5fc11492ef472dc5cfd2fd2c4926b1787;hpb=7a57cf92a463031c8cb668900f9215e3773a15c3 diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c index 76238a087..c2ad0fdd8 100644 --- a/src/common/ust-consumer/ust-consumer.c +++ b/src/common/ust-consumer/ust-consumer.c @@ -171,7 +171,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, case LTTNG_CONSUMER_ADD_STREAM: { struct lttng_consumer_stream *new_stream; - int fds[2]; + int fds[2], stream_pipe; size_t nb_fd = 2; struct consumer_relayd_sock_pair *relayd = NULL; int alloc_ret = 0; @@ -205,6 +205,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, msg.u.stream.gid, msg.u.stream.net_index, msg.u.stream.metadata_flag, + msg.u.stream.session_id, &alloc_ret); if (new_stream == NULL) { switch (alloc_ret) { @@ -234,12 +235,13 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, &new_stream->relayd_stream_id); pthread_mutex_unlock(&relayd->ctrl_sock_mutex); if (ret < 0) { + consumer_del_stream(new_stream, NULL); goto end_nosignal; } } else if (msg.u.stream.net_index != -1) { ERR("Network sequence index %d unknown. Not adding stream.", msg.u.stream.net_index); - free(new_stream); + consumer_del_stream(new_stream, NULL); goto end_nosignal; } @@ -247,24 +249,30 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, if (ctx->on_recv_stream) { ret = ctx->on_recv_stream(new_stream); if (ret < 0) { + consumer_del_stream(new_stream, NULL); goto end_nosignal; } } - /* Send stream to the metadata thread */ + /* Get the right pipe where the stream will be sent. */ if (new_stream->metadata_flag) { - do { - ret = write(ctx->consumer_metadata_pipe[1], &new_stream, - sizeof(new_stream)); - } while (ret < 0 && errno == EINTR); - if (ret < 0) { - PERROR("write metadata pipe"); - } + stream_pipe = ctx->consumer_metadata_pipe[1]; } else { - consumer_add_stream(new_stream); + stream_pipe = ctx->consumer_data_pipe[1]; + } + + do { + ret = write(stream_pipe, &new_stream, sizeof(new_stream)); + } while (ret < 0 && errno == EINTR); + if (ret < 0) { + PERROR("Consumer write %s stream to pipe %d", + new_stream->metadata_flag ? "metadata" : "data", + stream_pipe); + consumer_del_stream(new_stream, NULL); + goto end_nosignal; } - DBG("UST consumer_add_stream %s (%d,%d) with relayd id %" PRIu64, + DBG("UST consumer ADD_STREAM %s (%d,%d) with relayd id %" PRIu64, msg.u.stream.path_name, fds[0], fds[1], new_stream->relayd_stream_id); break; @@ -301,39 +309,27 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, { rcu_read_unlock(); return -ENOSYS; -#if 0 - if (ctx->on_update_stream != NULL) { - ret = ctx->on_update_stream(msg.u.stream.stream_key, msg.u.stream.state); - if (ret == 0) { - consumer_change_stream_state(msg.u.stream.stream_key, msg.u.stream.state); - } else if (ret < 0) { - goto end; - } - } else { - consumer_change_stream_state(msg.u.stream.stream_key, - msg.u.stream.state); + } + case LTTNG_CONSUMER_DATA_AVAILABLE: + { + int32_t ret; + uint64_t id = msg.u.data_available.session_id; + + DBG("UST consumer data available command for id %" PRIu64, id); + + ret = consumer_data_available(id); + + /* Send back returned value to session daemon */ + ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret)); + if (ret < 0) { + PERROR("send data available ret code"); } break; -#endif } default: break; } - /* - * Wake-up the other end by writing a null byte in the pipe (non-blocking). - * Important note: Because writing into the pipe is non-blocking (and - * therefore we allow dropping wakeup data, as long as there is wakeup data - * present in the pipe buffer to wake up the other end), the other end - * should perform the following sequence for waiting: - * - * 1) empty the pipe (reads). - * 2) perform update operation. - * 3) wait on the pipe (poll). - */ - do { - ret = write(ctx->consumer_poll_pipe[1], "", 1); - } while (ret < 0 && errno == EINTR); end_nosignal: rcu_read_unlock(); @@ -373,7 +369,7 @@ void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan) ustctl_unmap_channel(chan->handle); } -int lttng_ustconsumer_allocate_stream(struct lttng_consumer_stream *stream) +int lttng_ustconsumer_add_stream(struct lttng_consumer_stream *stream) { struct lttng_ust_object_data obj; int ret; @@ -385,13 +381,14 @@ int lttng_ustconsumer_allocate_stream(struct lttng_consumer_stream *stream) ret = ustctl_add_stream(stream->chan->handle, &obj); if (ret) { ERR("UST ctl add_stream failed with ret %d", ret); - return ret; + goto error; } stream->buf = ustctl_open_stream_read(stream->chan->handle, stream->cpu); if (!stream->buf) { ERR("UST ctl open_stream_read failed"); - return -EBUSY; + ret = -EBUSY; + goto error; } /* ustctl_open_stream_read has closed the shm fd. */ @@ -401,10 +398,16 @@ int lttng_ustconsumer_allocate_stream(struct lttng_consumer_stream *stream) stream->mmap_base = ustctl_get_mmap_base(stream->chan->handle, stream->buf); if (!stream->mmap_base) { ERR("UST ctl get_mmap_base failed"); - return -EINVAL; + ret = -EINVAL; + goto mmap_error; } return 0; + +mmap_error: + ustctl_close_stream_read(stream->chan->handle, stream->buf); +error: + return ret; } void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream) @@ -507,9 +510,58 @@ int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream) stream->out_fd = ret; } + ret = lttng_ustconsumer_add_stream(stream); + if (ret) { + consumer_del_stream(stream, NULL); + ret = -1; + goto error; + } + /* we return 0 to let the library handle the FD internally */ return 0; error: return ret; } + +/* + * Check if data is still being extracted from the buffers for a specific + * stream. Consumer data lock MUST be acquired before calling this function. + * + * Return 0 if the traced data are still getting read else 1 meaning that the + * data is available for trace viewer reading. + */ +int lttng_ustconsumer_data_available(struct lttng_consumer_stream *stream) +{ + int ret; + + assert(stream); + + DBG("UST consumer checking data availability"); + + /* + * Try to lock the stream mutex. On failure, we know that the stream is + * being used else where hence there is data still being extracted. + */ + ret = pthread_mutex_trylock(&stream->lock); + if (ret == EBUSY) { + goto data_not_available; + } + /* The stream is now locked so we can do our ustctl calls */ + + ret = ustctl_get_next_subbuf(stream->chan->handle, stream->buf); + if (ret == 0) { + /* There is still data so let's put back this subbuffer. */ + ret = ustctl_put_subbuf(stream->chan->handle, stream->buf); + assert(ret == 0); + pthread_mutex_unlock(&stream->lock); + goto data_not_available; + } + + /* Data is available to be read for this stream. */ + pthread_mutex_unlock(&stream->lock); + return 1; + +data_not_available: + return 0; +}