* Find a stream. The consumer_data.lock must be locked during this
* call.
*/
-static struct lttng_consumer_stream *consumer_find_stream(int key)
+static struct lttng_consumer_stream *consumer_find_stream(int key,
+ struct lttng_ht *ht)
{
struct lttng_ht_iter iter;
struct lttng_ht_node_ulong *node;
struct lttng_consumer_stream *stream = NULL;
+ assert(ht);
+
/* Negative keys are lookup failures */
- if (key < 0)
+ if (key < 0) {
return NULL;
+ }
rcu_read_lock();
- lttng_ht_lookup(consumer_data.stream_ht, (void *)((unsigned long) key),
- &iter);
+ lttng_ht_lookup(ht, (void *)((unsigned long) key), &iter);
node = lttng_ht_iter_get_node_ulong(&iter);
if (node != NULL) {
stream = caa_container_of(node, struct lttng_consumer_stream, node);
return stream;
}
-static void consumer_steal_stream_key(int key)
+static void consumer_steal_stream_key(int key, struct lttng_ht *ht)
{
struct lttng_consumer_stream *stream;
rcu_read_lock();
- stream = consumer_find_stream(key);
+ stream = consumer_find_stream(key, ht);
if (stream) {
stream->key = -1;
/*
struct lttng_consumer_channel *channel = NULL;
/* Negative keys are lookup failures */
- if (key < 0)
+ if (key < 0) {
return NULL;
+ }
rcu_read_lock();
if (stream->mmap_base != NULL) {
ret = munmap(stream->mmap_base, stream->mmap_len);
if (ret != 0) {
- perror("munmap");
+ PERROR("munmap");
}
}
break;
stream = zmalloc(sizeof(*stream));
if (stream == NULL) {
- perror("malloc struct lttng_consumer_stream");
+ PERROR("malloc struct lttng_consumer_stream");
*alloc_ret = -ENOMEM;
- return NULL;
+ goto end;
}
+
+ /*
+ * Get stream's channel reference. Needed when adding the stream to the
+ * global hash table.
+ */
stream->chan = consumer_find_channel(channel_key);
if (!stream->chan) {
*alloc_ret = -ENOENT;
+ ERR("Unable to find channel for stream %d", stream_key);
goto error;
}
stream->chan->refcount++;
error:
free(stream);
+end:
return NULL;
}
pthread_mutex_lock(&consumer_data.lock);
/* Steal stream identifier, for UST */
- consumer_steal_stream_key(stream->key);
+ consumer_steal_stream_key(stream->key, consumer_data.stream_ht);
rcu_read_lock();
lttng_ht_lookup(consumer_data.stream_ht,
struct lttng_consumer_stream *stream;
pthread_mutex_lock(&consumer_data.lock);
- stream = consumer_find_stream(stream_key);
+ stream = consumer_find_stream(stream_key, consumer_data.stream_ht);
if (stream) {
stream->state = state;
}
if (channel->mmap_base != NULL) {
ret = munmap(channel->mmap_base, channel->mmap_len);
if (ret != 0) {
- perror("munmap");
+ PERROR("munmap");
}
}
if (channel->wait_fd >= 0 && !channel->wait_fd_is_copy) {
channel = zmalloc(sizeof(*channel));
if (channel == NULL) {
- perror("malloc struct lttng_consumer_channel");
+ PERROR("malloc struct lttng_consumer_channel");
goto end;
}
channel->key = channel_key;
if (errno == EINTR) {
goto restart;
}
- perror("Poll error");
+ PERROR("Poll error");
goto exit;
}
if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) {
ret = write(ctx->consumer_should_quit[1], "4", 1);
} while (ret < 0 && errno == EINTR);
if (ret < 0) {
- perror("write consumer quit");
+ PERROR("write consumer quit");
}
}
ctx = zmalloc(sizeof(struct lttng_consumer_local_data));
if (ctx == NULL) {
- perror("allocating context");
+ PERROR("allocating context");
goto error;
}
ret = pipe(ctx->consumer_poll_pipe);
if (ret < 0) {
- perror("Error creating poll pipe");
+ PERROR("Error creating poll pipe");
goto error_poll_pipe;
}
/* set read end of the pipe to non-blocking */
ret = fcntl(ctx->consumer_poll_pipe[0], F_SETFL, O_NONBLOCK);
if (ret < 0) {
- perror("fcntl O_NONBLOCK");
+ PERROR("fcntl O_NONBLOCK");
goto error_poll_fcntl;
}
/* set write end of the pipe to non-blocking */
ret = fcntl(ctx->consumer_poll_pipe[1], F_SETFL, O_NONBLOCK);
if (ret < 0) {
- perror("fcntl O_NONBLOCK");
+ PERROR("fcntl O_NONBLOCK");
goto error_poll_fcntl;
}
ret = pipe(ctx->consumer_should_quit);
if (ret < 0) {
- perror("Error creating recv pipe");
+ PERROR("Error creating recv pipe");
goto error_quit_pipe;
}
ret = pipe(ctx->consumer_thread_pipe);
if (ret < 0) {
- perror("Error creating thread pipe");
+ PERROR("Error creating thread pipe");
goto error_thread_pipe;
}
}
/*
- * Iterate over all stream element of the hashtable and free them. This is race
- * free since the hashtable received MUST be in a race free synchronization
- * state. It's the caller responsability to make sure of that.
+ * Iterate over all streams of the hashtable and free them properly.
*/
static void destroy_stream_ht(struct lttng_ht *ht)
{
ret = lttng_ht_del(ht, &iter);
assert(!ret);
- free(stream);
+ call_rcu(&stream->node.head, consumer_free_stream);
}
rcu_read_unlock();
consumer_del_channel(stream->chan);
}
- free(stream);
+ call_rcu(&stream->node.head, consumer_free_stream);
}
/*
/* allocate for all fds + 1 for the consumer_poll_pipe */
pollfd = zmalloc((consumer_data.stream_count + 1) * sizeof(struct pollfd));
if (pollfd == NULL) {
- perror("pollfd malloc");
+ PERROR("pollfd malloc");
pthread_mutex_unlock(&consumer_data.lock);
goto end;
}
local_stream = zmalloc((consumer_data.stream_count + 1) *
sizeof(struct lttng_consumer_stream));
if (local_stream == NULL) {
- perror("local_stream malloc");
+ PERROR("local_stream malloc");
pthread_mutex_unlock(&consumer_data.lock);
goto end;
}
if (errno == EINTR) {
goto restart;
}
- perror("Poll error");
+ PERROR("Poll error");
lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
goto end;
} else if (num_rdy == 0) {
ret = fcntl(client_socket, F_SETFL, O_NONBLOCK);
if (ret < 0) {
- perror("fcntl O_NONBLOCK");
+ PERROR("fcntl O_NONBLOCK");
goto end;
}
}
ret = fcntl(sock, F_SETFL, O_NONBLOCK);
if (ret < 0) {
- perror("fcntl O_NONBLOCK");
+ PERROR("fcntl O_NONBLOCK");
goto end;
}
projects / lttng-tools.git / blobdiff