static struct lttng_ht *metadata_ht;
static struct lttng_ht *data_ht;
-/*
- * Notify a thread pipe to poll back again. This usually means that some global
- * state has changed so we just send back the thread in a poll wait call.
- */
-static void notify_thread_pipe(int wpipe)
-{
- int ret;
-
- do {
- struct lttng_consumer_stream *null_stream = NULL;
-
- ret = write(wpipe, &null_stream, sizeof(null_stream));
- } while (ret < 0 && errno == EINTR);
-}
-
/*
* Notify a thread lttng pipe to poll back again. This usually means that some
* global state has changed so we just send back the thread in a poll wait
msg.action = action;
msg.chan = chan;
+ msg.key = key;
do {
ret = write(ctx->consumer_channel_pipe[1], &msg, sizeof(msg));
} while (ret < 0 && errno == EINTR);
*/
if (ctx) {
notify_thread_lttng_pipe(ctx->consumer_data_pipe);
- notify_thread_pipe(ctx->consumer_metadata_pipe[1]);
+ notify_thread_lttng_pipe(ctx->consumer_metadata_pipe);
}
}
PERROR("munmap");
}
}
+
+ if (stream->wait_fd >= 0) {
+ ret = close(stream->wait_fd);
+ if (ret) {
+ PERROR("close");
+ }
+ }
break;
case LTTNG_CONSUMER32_UST:
case LTTNG_CONSUMER64_UST:
/* Channel already exist. Ignore the insertion */
ERR("Consumer add channel key %" PRIu64 " already exists!",
channel->key);
- ret = -1;
+ ret = LTTNG_ERR_KERN_CHAN_EXIST;
goto end;
}
goto error_channel_pipe;
}
- ret = utils_create_pipe(ctx->consumer_metadata_pipe);
- if (ret < 0) {
+ ctx->consumer_metadata_pipe = lttng_pipe_open(0);
+ if (!ctx->consumer_metadata_pipe) {
goto error_metadata_pipe;
}
return ctx;
error_splice_pipe:
- utils_close_pipe(ctx->consumer_metadata_pipe);
+ lttng_pipe_destroy(ctx->consumer_metadata_pipe);
error_metadata_pipe:
utils_close_pipe(ctx->consumer_channel_pipe);
error_channel_pipe:
utils_close_pipe(ctx->consumer_thread_pipe);
utils_close_pipe(ctx->consumer_channel_pipe);
lttng_pipe_destroy(ctx->consumer_data_pipe);
+ lttng_pipe_destroy(ctx->consumer_metadata_pipe);
utils_close_pipe(ctx->consumer_should_quit);
utils_close_pipe(ctx->consumer_splice_metadata_pipe);
PERROR("munmap metadata stream");
}
}
+
+ if (stream->wait_fd >= 0) {
+ ret = close(stream->wait_fd);
+ if (ret < 0) {
+ PERROR("close kernel metadata wait_fd");
+ }
+ }
break;
case LTTNG_CONSUMER32_UST:
case LTTNG_CONSUMER64_UST:
}
end:
+ /*
+ * Nullify the stream reference so it is not used after deletion. The
+ * consumer data lock MUST be acquired before being able to check for a
+ * NULL pointer value.
+ */
+ stream->chan->metadata_stream = NULL;
+
pthread_mutex_unlock(&stream->lock);
pthread_mutex_unlock(&consumer_data.lock);
uatomic_inc(&relayd->refcount);
}
- /* Update channel refcount once added without error(s). */
- uatomic_inc(&stream->chan->refcount);
-
/*
* When nb_init_stream_left reaches 0, we don't need to trigger any action
* in terms of destroying the associated channel, because the action that
goto end_poll;
}
- ret = lttng_poll_add(&events, ctx->consumer_metadata_pipe[0], LPOLLIN);
+ ret = lttng_poll_add(&events,
+ lttng_pipe_get_readfd(ctx->consumer_metadata_pipe), LPOLLIN);
if (ret < 0) {
goto end;
}
continue;
}
- if (pollfd == ctx->consumer_metadata_pipe[0]) {
+ if (pollfd == lttng_pipe_get_readfd(ctx->consumer_metadata_pipe)) {
if (revents & (LPOLLERR | LPOLLHUP )) {
DBG("Metadata thread pipe hung up");
/*
* Remove the pipe from the poll set and continue the loop
* since their might be data to consume.
*/
- lttng_poll_del(&events, ctx->consumer_metadata_pipe[0]);
- ret = close(ctx->consumer_metadata_pipe[0]);
- if (ret < 0) {
- PERROR("close metadata pipe");
- }
+ lttng_poll_del(&events,
+ lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
+ lttng_pipe_read_close(ctx->consumer_metadata_pipe);
continue;
} else if (revents & LPOLLIN) {
- do {
- /* Get the stream pointer received */
- ret = read(pollfd, &stream, sizeof(stream));
- } while (ret < 0 && errno == EINTR);
- if (ret < 0 ||
- ret < sizeof(struct lttng_consumer_stream *)) {
- PERROR("read metadata stream");
+ ssize_t pipe_len;
+
+ pipe_len = lttng_pipe_read(ctx->consumer_metadata_pipe,
+ &stream, sizeof(stream));
+ if (pipe_len < 0) {
+ ERR("read metadata stream, ret: %ld", pipe_len);
/*
- * Let's continue here and hope we can still work
- * without stopping the consumer. XXX: Should we?
+ * Continue here to handle the rest of the streams.
*/
continue;
}
* only tracked fd in the poll set. The thread will take care of closing
* the read side.
*/
- ret = close(ctx->consumer_metadata_pipe[1]);
- if (ret < 0) {
- PERROR("close data pipe");
- }
+ (void) lttng_pipe_write_close(ctx->consumer_metadata_pipe);
destroy_data_stream_ht(data_ht);
lttng_ht_node_init_u64(&chan->wait_fd_node,
chan->wait_fd);
+ rcu_read_lock();
lttng_ht_add_unique_u64(channel_ht,
&chan->wait_fd_node);
+ rcu_read_unlock();
/* Add channel to the global poll events list */
lttng_poll_add(&events, chan->wait_fd,
LPOLLIN | LPOLLPRI);
break;
case CONSUMER_CHANNEL_DEL:
{
+ rcu_read_lock();
chan = consumer_find_channel(key);
if (!chan) {
+ rcu_read_unlock();
ERR("UST consumer get channel key %" PRIu64 " not found for del channel", key);
break;
}
lttng_poll_del(&events, chan->wait_fd);
+ iter.iter.node = &chan->wait_fd_node.node;
ret = lttng_ht_del(channel_ht, &iter);
assert(ret == 0);
consumer_close_channel_streams(chan);
if (!uatomic_sub_return(&chan->refcount, 1)) {
consumer_del_channel(chan);
}
+ rcu_read_unlock();
goto restart;
}
case CONSUMER_CHANNEL_QUIT:
}
}
if (client_socket >= 0) {
- ret = close(sock);
+ ret = close(client_socket);
if (ret < 0) {
PERROR("close client_socket sessiond poll");
}
projects / lttng-tools.git / blobdiff