Map session id of relayd and sessiond in consumer

Acked-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Signed-off-by: David Goulet <dgoulet@efficios.com>

diff --git a/src/bin/lttng-sessiond/cmd.c b/src/bin/lttng-sessiond/cmd.c
index aaa5a8923332bb39ed3b43a8ecdc5688880206dd..491acde45f760a5613c338a91263231cffffbbc3 100644 (file)
--- a/src/bin/lttng-sessiond/cmd.c
+++ b/src/bin/lttng-sessiond/cmd.c
@@ -558,7 +558,7 @@ static int send_consumer_relayd_socket(int domain, struct ltt_session *session,
 
        /* Send relayd socket to consumer. */
        ret = consumer_send_relayd_socket(consumer_sock, sock,
 
        /* Send relayd socket to consumer. */
        ret = consumer_send_relayd_socket(consumer_sock, sock,

-                       consumer, relayd_uri->stype);
+                       consumer, relayd_uri->stype, session->id);

        if (ret < 0) {
                ret = LTTNG_ERR_ENABLE_CONSUMER_FAIL;
                goto close_sock;
        if (ret < 0) {
                ret = LTTNG_ERR_ENABLE_CONSUMER_FAIL;
                goto close_sock;

diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c
index 92a6c5ddc72a0fca50f40fd2c1d03a0b7cddde12..fd9e1e6a0667f3e55f7946540883fb1277313129 100644 (file)
--- a/src/bin/lttng-sessiond/consumer.c
+++ b/src/bin/lttng-sessiond/consumer.c
@@ -658,7 +658,7 @@ error:
  */
 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
                struct lttcomm_sock *sock, struct consumer_output *consumer,
  */
 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
                struct lttcomm_sock *sock, struct consumer_output *consumer,

-               enum lttng_stream_type type)
+               enum lttng_stream_type type, unsigned int session_id)

 {
        int ret;
        struct lttcomm_consumer_msg msg;
 {
        int ret;
        struct lttcomm_consumer_msg msg;


@@ -682,6 +682,7 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
         */
        msg.u.relayd_sock.net_index = consumer->net_seq_index;
        msg.u.relayd_sock.type = type;
         */
        msg.u.relayd_sock.net_index = consumer->net_seq_index;
        msg.u.relayd_sock.type = type;

+       msg.u.relayd_sock.session_id = session_id;

        memcpy(&msg.u.relayd_sock.sock, sock, sizeof(msg.u.relayd_sock.sock));
 
        DBG3("Sending relayd sock info to consumer on %d", consumer_sock->fd);
        memcpy(&msg.u.relayd_sock.sock, sock, sizeof(msg.u.relayd_sock.sock));
 
        DBG3("Sending relayd sock info to consumer on %d", consumer_sock->fd);

diff --git a/src/bin/lttng-sessiond/consumer.h b/src/bin/lttng-sessiond/consumer.h
index 01548cd17f088bfab82e6bd57082ed2ff87c1510..1e7e3b1d579cc1287ac20abd45ccf35d5a70875a 100644 (file)
--- a/src/bin/lttng-sessiond/consumer.h
+++ b/src/bin/lttng-sessiond/consumer.h
@@ -171,7 +171,7 @@ int consumer_send_channel(struct consumer_socket *sock,
                struct lttcomm_consumer_msg *msg);
 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
                struct lttcomm_sock *sock, struct consumer_output *consumer,
                struct lttcomm_consumer_msg *msg);
 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
                struct lttcomm_sock *sock, struct consumer_output *consumer,

-               enum lttng_stream_type type);
+               enum lttng_stream_type type, unsigned int session_id);

 int consumer_send_destroy_relayd(struct consumer_socket *sock,
                struct consumer_output *consumer);
 int consumer_recv_status_reply(struct consumer_socket *sock);
 int consumer_send_destroy_relayd(struct consumer_socket *sock,
                struct consumer_output *consumer);
 int consumer_recv_status_reply(struct consumer_socket *sock);

diff --git a/src/common/consumer.c b/src/common/consumer.c
index f4cfa82ce8dcc1179843a806a4d969f432b5391e..b5e212f27d61c5f08f53ef32fb19ac91c1e55917 100644 (file)
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -63,6 +63,20 @@ volatile int consumer_quit;
 static struct lttng_ht *metadata_ht;
 static struct lttng_ht *data_ht;
 
 static struct lttng_ht *metadata_ht;
 static struct lttng_ht *data_ht;
 

+/*
+ * This hash table contains the mapping between the session id of the sessiond
+ * and the relayd session id. Element of the ht are indexed by sessiond_id.
+ *
+ * Node can be added when a relayd communication is opened in the sessiond
+ * thread.
+ *
+ * Note that a session id of the session daemon is unique to a tracing session
+ * and not to a domain session. However, a domain session has one consumer
+ * which forces the 1-1 mapping between a consumer and a domain session (ex:
+ * UST). This means that we can't have duplicate in this ht.
+ */
+static struct lttng_ht *relayd_session_id_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.
 /*
  * 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.


@@ -216,6 +230,7 @@ static void destroy_relayd(struct consumer_relayd_sock_pair *relayd)
 {
        int ret;
        struct lttng_ht_iter iter;
 {
        int ret;
        struct lttng_ht_iter iter;

+       struct lttng_ht_node_ulong *node;

 
        if (relayd == NULL) {
                return;
 
        if (relayd == NULL) {
                return;


@@ -223,6 +238,20 @@ static void destroy_relayd(struct consumer_relayd_sock_pair *relayd)
 
        DBG("Consumer destroy and close relayd socket pair");
 
 
        DBG("Consumer destroy and close relayd socket pair");
 

+       lttng_ht_lookup(relayd_session_id_ht,
+                       (void *)((unsigned long) relayd->session_id), &iter);
+       node = lttng_ht_iter_get_node_ulong(&iter);
+       if (node != NULL) {
+               /* We assume the relayd is being or is destroyed */
+               return;
+       }
+
+       ret = lttng_ht_del(relayd_session_id_ht, &iter);
+       if (ret != 0) {
+               /* We assume the relayd is being or is destroyed */
+               return;
+       }
+

        iter.iter.node = &relayd->node.node;
        ret = lttng_ht_del(consumer_data.relayd_ht, &iter);
        if (ret != 0) {
        iter.iter.node = &relayd->node.node;
        ret = lttng_ht_del(consumer_data.relayd_ht, &iter);
        if (ret != 0) {


@@ -2666,6 +2695,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);
        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);

+       relayd_session_id_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);

 }
 
 /*
 }
 
 /*


@@ -2676,11 +2706,13 @@ void lttng_consumer_init(void)
  */
 int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
                struct lttng_consumer_local_data *ctx, int sock,
  */
 int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
                struct lttng_consumer_local_data *ctx, int sock,

-               struct pollfd *consumer_sockpoll, struct lttcomm_sock *relayd_sock)
+               struct pollfd *consumer_sockpoll, struct lttcomm_sock *relayd_sock,
+               unsigned int sessiond_id)

 {
        int fd = -1, ret = -1;
        enum lttng_error_code ret_code = LTTNG_OK;
        struct consumer_relayd_sock_pair *relayd;
 {
        int fd = -1, ret = -1;
        enum lttng_error_code ret_code = LTTNG_OK;
        struct consumer_relayd_sock_pair *relayd;

+       struct consumer_relayd_session_id *relayd_id_node;

 
        DBG("Consumer adding relayd socket (idx: %d)", net_seq_idx);
 
 
        DBG("Consumer adding relayd socket (idx: %d)", net_seq_idx);
 


@@ -2754,6 +2786,23 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
                        goto error;
                }
 
                        goto error;
                }
 

+               /* Set up a relayd session id node. */
+               relayd_id_node = zmalloc(sizeof(struct consumer_relayd_session_id));
+               if (!relayd_id_node) {
+                       PERROR("zmalloc relayd id node");
+                       goto error;
+               }
+
+               relayd_id_node->relayd_id = relayd->session_id;
+               relayd_id_node->sessiond_id = (uint64_t) sessiond_id;
+
+               /* Indexed by session id of the session daemon. */
+               lttng_ht_node_init_ulong(&relayd_id_node->node,
+                               relayd_id_node->sessiond_id);
+               rcu_read_lock();
+               lttng_ht_add_unique_ulong(relayd_session_id_ht, &relayd_id_node->node);
+               rcu_read_unlock();
+

                break;
        case LTTNG_STREAM_DATA:
                /* Copy received lttcomm socket */
                break;
        case LTTNG_STREAM_DATA:
                /* Copy received lttcomm socket */

diff --git a/src/common/consumer.h b/src/common/consumer.h
index 80bb46a830ab1f9b9649d4d4ecee9cf7a469ddb5..e9927c7d46556150eab2520af8fdf01f72c77e6e 100644 (file)
--- a/src/common/consumer.h
+++ b/src/common/consumer.h
@@ -299,6 +299,15 @@ struct lttng_consumer_global_data {
        struct lttng_ht *stream_list_ht;
 };
 
        struct lttng_ht *stream_list_ht;
 };
 

+/*
+ * Session id mapping structure storred in relayd_session_id_ht.
+ */
+struct consumer_relayd_session_id {
+       uint64_t sessiond_id;
+       uint64_t relayd_id;
+       struct lttng_ht_node_ulong node;
+};
+

 /*
  * Init consumer data structures.
  */
 /*
  * Init consumer data structures.
  */


@@ -415,7 +424,8 @@ ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream);
 int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
                struct lttng_consumer_local_data *ctx, int sock,
 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream);
 int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
                struct lttng_consumer_local_data *ctx, int sock,

-               struct pollfd *consumer_sockpoll, struct lttcomm_sock *relayd_sock);
+               struct pollfd *consumer_sockpoll, struct lttcomm_sock *relayd_sock,
+               unsigned int sessiond_id);

 void consumer_flag_relayd_for_destroy(
                struct consumer_relayd_sock_pair *relayd);
 int consumer_data_pending(uint64_t id);
 void consumer_flag_relayd_for_destroy(
                struct consumer_relayd_sock_pair *relayd);
 int consumer_data_pending(uint64_t id);

diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c
index 40d0f71960fb6204019b508bd0ca1caffb4f838c..bc4f59716a18edf92d70f5b1de43430d7c011824 100644 (file)
--- a/src/common/kernel-consumer/kernel-consumer.c
+++ b/src/common/kernel-consumer/kernel-consumer.c
@@ -117,7 +117,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                /* Session daemon status message are handled in the following call. */
                ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
                                msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
                /* Session daemon status message are handled in the following call. */
                ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
                                msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,

-                               &msg.u.relayd_sock.sock);
+                               &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id);

                goto end_nosignal;
        }
        case LTTNG_CONSUMER_ADD_CHANNEL:
                goto end_nosignal;
        }
        case LTTNG_CONSUMER_ADD_CHANNEL:

diff --git a/src/common/sessiond-comm/sessiond-comm.h b/src/common/sessiond-comm/sessiond-comm.h
index 4dd5731221e929f2f1e94984ce17826327797866..2cd1d5407ea06b09d9fbbf685e973e0a08e42b9d 100644 (file)
--- a/src/common/sessiond-comm/sessiond-comm.h
+++ b/src/common/sessiond-comm/sessiond-comm.h
@@ -281,6 +281,8 @@ struct lttcomm_consumer_msg {
                        enum lttng_stream_type type;
                        /* Open socket to the relayd */
                        struct lttcomm_sock sock;
                        enum lttng_stream_type type;
                        /* Open socket to the relayd */
                        struct lttcomm_sock sock;

+                       /* Tracing session id associated to the relayd. */
+                       uint64_t session_id;

                } relayd_sock;
                struct {
                        uint64_t net_seq_idx;
                } relayd_sock;
                struct {
                        uint64_t net_seq_idx;

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index c0999b613b496446c3c0de472af7f76cad6085d1..b49f59cb061049f45a0ff78c6a6a08f1e41fb564 100644 (file)
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -132,7 +132,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                /* Session daemon status message are handled in the following call. */
                ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
                                msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
                /* Session daemon status message are handled in the following call. */
                ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
                                msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,

-                               &msg.u.relayd_sock.sock);
+                               &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id);

                goto end_nosignal;
        }
        case LTTNG_CONSUMER_ADD_CHANNEL:
                goto end_nosignal;
        }
        case LTTNG_CONSUMER_ADD_CHANNEL: