consumerd: pass channel instance to stream creation function

Both callsites of consumer_allocate_stream() set the stream's "chan"
pointer after the creation. Pass the channel directly to the stream
creation function so it can initialize the stream according to the
channel's settings.

Signed-off-by: Jérémie Galarneau <jeremie.galarneau@efficios.com>
Change-Id: Icea7088e7695e310585bf398e14e6443d67a30bb

diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c
index f06ea9d4d3cda1d23f88b63136fc323e45c0e924..e1ee5f2fc7d411689b2c5c28949abbe8aa383c82 100644 (file)
--- a/src/common/consumer/consumer.c
+++ b/src/common/consumer/consumer.c
@@ -532,7 +532,9 @@ void consumer_del_stream_for_metadata(struct lttng_consumer_stream *stream)
        consumer_stream_destroy(stream, metadata_ht);
 }
 
-struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
+struct lttng_consumer_stream *consumer_allocate_stream(
+               struct lttng_consumer_channel *channel,
+               uint64_t channel_key,
                uint64_t stream_key,
                enum lttng_consumer_stream_state state,
                const char *channel_name,
@@ -556,7 +558,7 @@ struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
        }
 
        rcu_read_lock();
-
+       stream->chan = channel;
        stream->key = stream_key;
        stream->out_fd = -1;
        stream->out_fd_offset = 0;

diff --git a/src/common/consumer/consumer.h b/src/common/consumer/consumer.h
index aebcf4baccb9711eb36fa1521c2ae857e7ee8b34..9ba1dacbc6f52b9cf1b9caa4d1880fd9e93fd308 100644 (file)
--- a/src/common/consumer/consumer.h
+++ b/src/common/consumer/consumer.h
@@ -645,7 +645,9 @@ void lttng_consumer_cleanup(void);
  */
 int lttng_consumer_poll_socket(struct pollfd *kconsumer_sockpoll);
 
-struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
+struct lttng_consumer_stream *consumer_allocate_stream(
+               struct lttng_consumer_channel *channel,
+               uint64_t channel_key,
                uint64_t stream_key,
                enum lttng_consumer_stream_state state,
                const char *channel_name,

diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c
index 49b5070f76171b4613922bd84c1f4b7f7b941193..29b75d68d00d12d96cb62cfaf9ad3477d3f1314c 100644 (file)
--- a/src/common/kernel-consumer/kernel-consumer.c
+++ b/src/common/kernel-consumer/kernel-consumer.c
@@ -650,7 +650,10 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 
                health_code_update();
 
-               new_stream = consumer_allocate_stream(channel->key,
+               pthread_mutex_lock(&channel->lock);
+               new_stream = consumer_allocate_stream(
+                               channel,
+                               channel->key,
                                fd,
                                LTTNG_CONSUMER_ACTIVE_STREAM,
                                channel->name,
@@ -670,10 +673,10 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                                lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
                                break;
                        }
+                       pthread_mutex_unlock(&channel->lock);
                        goto end_nosignal;
                }
 
-               new_stream->chan = channel;
                new_stream->wait_fd = fd;
                switch (channel->output) {
                case CONSUMER_CHANNEL_SPLICE:
@@ -733,6 +736,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                                        "relayd id %" PRIu64, new_stream->name,
                                        new_stream->relayd_id);
                        cds_list_add(&new_stream->send_node, &channel->streams.head);
+                       pthread_mutex_unlock(&channel->lock);
                        break;
                }
 
@@ -741,6 +745,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                        ret = consumer_send_relayd_stream(new_stream,
                                        new_stream->chan->pathname);
                        if (ret < 0) {
+                               pthread_mutex_unlock(&channel->lock);
                                consumer_stream_free(new_stream);
                                goto end_nosignal;
                        }
@@ -754,10 +759,12 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                                ret = consumer_send_relayd_streams_sent(
                                                new_stream->relayd_id);
                                if (ret < 0) {
+                                       pthread_mutex_unlock(&channel->lock);
                                        goto end_nosignal;
                                }
                        }
                }
+               pthread_mutex_unlock(&channel->lock);
 
                /* Get the right pipe where the stream will be sent. */
                if (new_stream->metadata_flag) {

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 0318eecea05600ded402806ef24d6d32f7a7b23e..c37be8fb8fa49f103d04125861a6465bb7667575 100644 (file)
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -155,7 +155,9 @@ static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
        assert(channel);
        assert(ctx);
 
-       stream = consumer_allocate_stream(channel->key,
+       stream = consumer_allocate_stream(
+                       channel,
+                       channel->key,
                        key,
                        LTTNG_CONSUMER_ACTIVE_STREAM,
                        channel->name,
@@ -185,8 +187,6 @@ static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
                goto error;
        }
 
-       stream->chan = channel;
-
 error:
        if (_alloc_ret) {
                *_alloc_ret = alloc_ret;