enum consumer_channel_action {
CONSUMER_CHANNEL_ADD,
+ CONSUMER_CHANNEL_DEL,
CONSUMER_CHANNEL_QUIT,
};
struct consumer_channel_msg {
enum consumer_channel_action action;
- struct lttng_consumer_channel *chan;
+ struct lttng_consumer_channel *chan; /* add */
+ uint64_t key; /* del */
};
/*
} 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
+ * call.
+ */
+static void notify_thread_lttng_pipe(struct lttng_pipe *pipe)
+{
+ struct lttng_consumer_stream *null_stream = NULL;
+
+ assert(pipe);
+
+ (void) lttng_pipe_write(pipe, &null_stream, sizeof(null_stream));
+}
+
static void notify_channel_pipe(struct lttng_consumer_local_data *ctx,
struct lttng_consumer_channel *chan,
+ uint64_t key,
enum consumer_channel_action action)
{
struct consumer_channel_msg msg;
int ret;
+ memset(&msg, 0, sizeof(msg));
+
msg.action = action;
msg.chan = chan;
do {
} while (ret < 0 && errno == EINTR);
}
+void notify_thread_del_channel(struct lttng_consumer_local_data *ctx,
+ uint64_t key)
+{
+ notify_channel_pipe(ctx, NULL, key, CONSUMER_CHANNEL_DEL);
+}
+
static int read_channel_pipe(struct lttng_consumer_local_data *ctx,
struct lttng_consumer_channel **chan,
+ uint64_t *key,
enum consumer_channel_action *action)
{
struct consumer_channel_msg msg;
if (ret > 0) {
*action = msg.action;
*chan = msg.chan;
+ *key = msg.key;
}
return ret;
}
* read of this status which happens AFTER receiving this notify.
*/
if (ctx) {
- notify_thread_pipe(ctx->consumer_data_pipe[1]);
+ notify_thread_lttng_pipe(ctx->consumer_data_pipe);
notify_thread_pipe(ctx->consumer_metadata_pipe[1]);
}
}
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
if (!ret && channel->wait_fd != -1 &&
channel->metadata_stream == NULL) {
- notify_channel_pipe(ctx, channel, CONSUMER_CHANNEL_ADD);
+ notify_channel_pipe(ctx, channel, -1, CONSUMER_CHANNEL_ADD);
}
return ret;
}
* Insert the consumer_data_pipe at the end of the array and don't
* increment i so nb_fd is the number of real FD.
*/
- (*pollfd)[i].fd = ctx->consumer_data_pipe[0];
+ (*pollfd)[i].fd = lttng_pipe_get_readfd(ctx->consumer_data_pipe);
(*pollfd)[i].events = POLLIN | POLLPRI;
return i;
}
ctx->on_recv_stream = recv_stream;
ctx->on_update_stream = update_stream;
- ret = pipe(ctx->consumer_data_pipe);
- if (ret < 0) {
- PERROR("Error creating poll pipe");
+ ctx->consumer_data_pipe = lttng_pipe_open(0);
+ if (!ctx->consumer_data_pipe) {
goto error_poll_pipe;
}
- /* set read end of the pipe to non-blocking */
- ret = fcntl(ctx->consumer_data_pipe[0], F_SETFL, O_NONBLOCK);
- if (ret < 0) {
- PERROR("fcntl O_NONBLOCK");
- goto error_poll_fcntl;
- }
-
- /* set write end of the pipe to non-blocking */
- ret = fcntl(ctx->consumer_data_pipe[1], F_SETFL, O_NONBLOCK);
- if (ret < 0) {
- PERROR("fcntl O_NONBLOCK");
- goto error_poll_fcntl;
- }
-
ret = pipe(ctx->consumer_should_quit);
if (ret < 0) {
PERROR("Error creating recv pipe");
utils_close_pipe(ctx->consumer_thread_pipe);
error_thread_pipe:
utils_close_pipe(ctx->consumer_should_quit);
-error_poll_fcntl:
error_quit_pipe:
- utils_close_pipe(ctx->consumer_data_pipe);
+ lttng_pipe_destroy(ctx->consumer_data_pipe);
error_poll_pipe:
free(ctx);
error:
}
utils_close_pipe(ctx->consumer_thread_pipe);
utils_close_pipe(ctx->consumer_channel_pipe);
- utils_close_pipe(ctx->consumer_data_pipe);
+ lttng_pipe_destroy(ctx->consumer_data_pipe);
utils_close_pipe(ctx->consumer_should_quit);
utils_close_pipe(ctx->consumer_splice_metadata_pipe);
goto end;
}
outfd = stream->out_fd = ret;
+ /* Reset current size because we just perform a rotation. */
+ stream->tracefile_size_current = 0;
}
stream->tracefile_size_current += len;
}
goto end;
}
outfd = stream->out_fd = ret;
+ /* Reset current size because we just perform a rotation. */
+ stream->tracefile_size_current = 0;
}
stream->tracefile_size_current += len;
}
ssize_t pipe_readlen;
DBG("consumer_data_pipe wake up");
- /* Consume 1 byte of pipe data */
- do {
- pipe_readlen = read(ctx->consumer_data_pipe[0], &new_stream,
- sizeof(new_stream));
- } while (pipe_readlen == -1 && errno == EINTR);
+ pipe_readlen = lttng_pipe_read(ctx->consumer_data_pipe,
+ &new_stream, sizeof(new_stream));
if (pipe_readlen < 0) {
- PERROR("read consumer data pipe");
+ ERR("Consumer data pipe ret %ld", pipe_readlen);
/* Continue so we can at least handle the current stream(s). */
continue;
}
continue;
} else if (revents & LPOLLIN) {
enum consumer_channel_action action;
+ uint64_t key;
- ret = read_channel_pipe(ctx, &chan, &action);
+ ret = read_channel_pipe(ctx, &chan, &key, &action);
if (ret <= 0) {
ERR("Error reading channel pipe");
continue;
lttng_poll_add(&events, chan->wait_fd,
LPOLLIN | LPOLLPRI);
break;
+ case CONSUMER_CHANNEL_DEL:
+ {
+ chan = consumer_find_channel(key);
+ if (!chan) {
+ ERR("UST consumer get channel key %" PRIu64 " not found for del channel", key);
+ break;
+ }
+ lttng_poll_del(&events, chan->wait_fd);
+ ret = lttng_ht_del(channel_ht, &iter);
+ assert(ret == 0);
+ consumer_close_channel_streams(chan);
+
+ /*
+ * Release our own refcount. Force channel deletion even if
+ * streams were not initialized.
+ */
+ if (!uatomic_sub_return(&chan->refcount, 1)) {
+ consumer_del_channel(chan);
+ }
+ goto restart;
+ }
case CONSUMER_CHANNEL_QUIT:
/*
* Remove the pipe from the poll set and continue the loop
* Notify the data poll thread to poll back again and test the
* consumer_quit state that we just set so to quit gracefully.
*/
- notify_thread_pipe(ctx->consumer_data_pipe[1]);
+ notify_thread_lttng_pipe(ctx->consumer_data_pipe);
- notify_channel_pipe(ctx, NULL, CONSUMER_CHANNEL_QUIT);
+ notify_channel_pipe(ctx, NULL, -1, CONSUMER_CHANNEL_QUIT);
/* Cleaning up possibly open sockets. */
if (sock >= 0) {