Add data structure for the data available command

The new stream hash table indexed by session id is added. The per
consumer stream mutex used for the synchronization between threads
accessing the stream which will be indexed in two hash tables.

Also, the data available command is added to both UST and kernel
consumer but for now returns ENOSYS. Just set the skeleton for the
implementation.

Finally, the session id is added to the trace-kernel (where it already
exists for UST) so it could be used later on for the data available
command.

This session id is also added in the consumer data structure for the
same purpose.

Signed-off-by: David Goulet <dgoulet@efficios.com>

diff --git a/src/bin/lttng-sessiond/kernel.c b/src/bin/lttng-sessiond/kernel.c
index 34695efff568e3855e8df32f79af4a895ecc97a0..bf03dbad985388a2355dcb2a46b3ad7281c9fccf 100644 (file)
--- a/src/bin/lttng-sessiond/kernel.c
+++ b/src/bin/lttng-sessiond/kernel.c
@@ -125,6 +125,7 @@ int kernel_create_session(struct ltt_session *session, int tracer_fd)
                PERROR("fcntl session fd");
        }
 
+       lks->id = session->id;
        lks->consumer_fds_sent = 0;
        session->kernel_session = lks;
 

diff --git a/src/bin/lttng-sessiond/trace-kernel.h b/src/bin/lttng-sessiond/trace-kernel.h
index c86cc273ce8ba363865a706938a913687fd7dfab..e5a8f3b0b4c0f6acd7268cf760bca94e777b6503 100644 (file)
--- a/src/bin/lttng-sessiond/trace-kernel.h
+++ b/src/bin/lttng-sessiond/trace-kernel.h
@@ -107,6 +107,8 @@ struct ltt_kernel_session {
         */
        struct consumer_output *consumer;
        struct consumer_output *tmp_consumer;
+       /* Tracing session id */
+       unsigned int id;
 };
 
 /*

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 0f0e60ad839a9beab57168634a8d7220a86fae62..464198597b8cc9ece7ee00d465077cbe757eb00c 100644 (file)
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -364,6 +364,7 @@ struct lttng_consumer_stream *consumer_allocate_stream(
                gid_t gid,
                int net_index,
                int metadata_flag,
+               uint64_t session_id,
                int *alloc_ret)
 {
        struct lttng_consumer_stream *stream;
@@ -399,8 +400,10 @@ struct lttng_consumer_stream *consumer_allocate_stream(
        stream->gid = gid;
        stream->net_seq_idx = net_index;
        stream->metadata_flag = metadata_flag;
+       stream->session_id = session_id;
        strncpy(stream->path_name, path_name, sizeof(stream->path_name));
        stream->path_name[sizeof(stream->path_name) - 1] = '\0';
+       pthread_mutex_init(&stream->lock, NULL);
 
        /*
         * Index differently the metadata node because the thread is using an
@@ -413,6 +416,9 @@ struct lttng_consumer_stream *consumer_allocate_stream(
                lttng_ht_node_init_ulong(&stream->node, stream->key);
        }
 
+       /* Init session id node with the stream session id */
+       lttng_ht_node_init_ulong(&stream->node_session_id, stream->session_id);
+
        /*
         * The cpu number is needed before using any ustctl_* actions. Ignored for
         * the kernel so the value does not matter.
@@ -422,10 +428,10 @@ struct lttng_consumer_stream *consumer_allocate_stream(
        pthread_mutex_unlock(&consumer_data.lock);
 
        DBG3("Allocated stream %s (key %d, shm_fd %d, wait_fd %d, mmap_len %llu,"
-                       " out_fd %d, net_seq_idx %d)", stream->path_name, stream->key,
-                       stream->shm_fd, stream->wait_fd,
+                       " out_fd %d, net_seq_idx %d, session_id %" PRIu64,
+                       stream->path_name, stream->key, stream->shm_fd, stream->wait_fd,
                        (unsigned long long) stream->mmap_len, stream->out_fd,
-                       stream->net_seq_idx);
+                       stream->net_seq_idx, stream->session_id);
        return stream;
 
 error:
@@ -2308,6 +2314,7 @@ void lttng_consumer_init(void)
 {
        consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
        consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
+       consumer_data.stream_list_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
 
        metadata_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
        assert(metadata_ht);

diff --git a/src/common/consumer.h b/src/common/consumer.h
index fe1568382f3971728fb2a15fbc4b7bec348541d5..9981856feeca8734a428951418c34c7e2f380e96 100644 (file)
--- a/src/common/consumer.h
+++ b/src/common/consumer.h
@@ -56,6 +56,8 @@ enum lttng_consumer_command {
        LTTNG_CONSUMER_ADD_RELAYD_SOCKET,
        /* Inform the consumer to kill a specific relayd connection */
        LTTNG_CONSUMER_DESTROY_RELAYD,
+       /* Return to the sessiond if there is data pending for a session */
+       LTTNG_CONSUMER_DATA_AVAILABLE,
 };
 
 /* State of each fd in consumer */
@@ -101,8 +103,10 @@ struct lttng_ust_lib_ring_buffer;
  * uniquely a stream.
  */
 struct lttng_consumer_stream {
-       /* Hash table node for both metadata and data type */
+       /* HT node used by the data_ht and metadata_ht */
        struct lttng_ht_node_ulong node;
+       /* HT node used in consumer_data.stream_list_ht */
+       struct lttng_ht_node_ulong node_session_id;
        struct lttng_consumer_channel *chan;    /* associated channel */
        /*
         * key is the key used by the session daemon to refer to the
@@ -137,6 +141,10 @@ struct lttng_consumer_stream {
        uint64_t relayd_stream_id;
        /* Next sequence number to use for trace packet */
        uint64_t next_net_seq_num;
+       /* Lock to use the stream FDs since they are used between threads. */
+       pthread_mutex_t lock;
+       /* Tracing session id */
+       uint64_t session_id;
 };
 
 /*
@@ -272,6 +280,15 @@ struct lttng_consumer_global_data {
         * stream has an index which associate the right relayd socket to use.
         */
        struct lttng_ht *relayd_ht;
+
+       /*
+        * This hash table contains all streams (metadata and data) indexed by
+        * session id. In other words, the ht is indexed by session id and each
+        * bucket contains the list of associated streams.
+        *
+        * This HT uses the "node_session_id" of the consumer stream.
+        */
+       struct lttng_ht *stream_list_ht;
 };
 
 /* Defined in consumer.c and coupled with explanations */
@@ -338,6 +355,7 @@ extern struct lttng_consumer_stream *consumer_allocate_stream(
                gid_t gid,
                int net_index,
                int metadata_flag,
+               uint64_t session_id,
                int *alloc_ret);
 extern void consumer_del_stream(struct lttng_consumer_stream *stream,
                struct lttng_ht *ht);

diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c
index c762934ff92a55569b9323852589157e3e4ceb81..2456d3fc91936c69f074bbd038ee227c46e48220 100644 (file)
--- a/src/common/kernel-consumer/kernel-consumer.c
+++ b/src/common/kernel-consumer/kernel-consumer.c
@@ -168,6 +168,7 @@ int lttng_kconsumer_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) {
@@ -279,6 +280,11 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 
                goto end_nosignal;
        }
+       case LTTNG_CONSUMER_DATA_AVAILABLE:
+       {
+               rcu_read_unlock();
+               return -ENOSYS;
+       }
        default:
                goto end_nosignal;
        }

diff --git a/src/common/sessiond-comm/sessiond-comm.h b/src/common/sessiond-comm/sessiond-comm.h
index 753e842340e5f4fd882ca99a5d3e4e1c151725a6..5d2fa36a8b02fa9b7417b92c87a49b5a742c384f 100644 (file)
--- a/src/common/sessiond-comm/sessiond-comm.h
+++ b/src/common/sessiond-comm/sessiond-comm.h
@@ -272,6 +272,7 @@ struct lttcomm_consumer_msg {
                        int net_index;
                        unsigned int metadata_flag;
                        char name[LTTNG_SYMBOL_NAME_LEN];  /* Name string of the stream */
+                       uint64_t session_id;   /* Tracing session id of the stream */
                } stream;
                struct {
                        int net_index;
@@ -282,6 +283,9 @@ struct lttcomm_consumer_msg {
                struct {
                        uint64_t net_seq_idx;
                } destroy_relayd;
+               struct {
+                       uint64_t session_id;
+               } data_available;
        } u;
 };
 

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index dea92ac2374f3f5127dc97389ed6f054d8f49d69..f802c462d4393565d5c3da60be70bd6b8d1cad3e 100644 (file)
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -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) {
@@ -309,6 +310,11 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                rcu_read_unlock();
                return -ENOSYS;
        }
+       case LTTNG_CONSUMER_DATA_AVAILABLE:
+       {
+               rcu_read_unlock();
+               return -ENOSYS;
+       }
        default:
                break;
        }