Fix: set relayd sock sent flag per consumer socket

[lttng-tools.git] / src / bin / lttng-sessiond / cmd.c

diff --git a/src/bin/lttng-sessiond/cmd.c b/src/bin/lttng-sessiond/cmd.c
index 2110e31e9508683f7332088c20e147bf84e3c921..0f94a5f36b695b0bebc2e4b58cbf89a060a2f322 100644 (file)
--- a/src/bin/lttng-sessiond/cmd.c
+++ b/src/bin/lttng-sessiond/cmd.c
@@ -152,6 +152,7 @@ static void list_lttng_channels(int domain, struct ltt_session *session,
                struct lttng_ht_iter iter;
                struct ltt_ust_channel *uchan;
 
+               rcu_read_lock();
                cds_lfht_for_each_entry(session->ust_session->domain_global.channels->ht,
                                &iter.iter, uchan, node.node) {
                        strncpy(channels[i].name, uchan->name, LTTNG_SYMBOL_NAME_LEN);
@@ -171,6 +172,7 @@ static void list_lttng_channels(int domain, struct ltt_session *session,
                        }
                        i++;
                }
+               rcu_read_unlock();
                break;
        }
        default:
@@ -433,6 +435,8 @@ static int init_kernel_tracing(struct ltt_kernel_session *session)
 
        assert(session);
 
+       rcu_read_lock();
+
        if (session->consumer_fds_sent == 0 && session->consumer != NULL) {
                cds_lfht_for_each_entry(session->consumer->socks->ht, &iter.iter,
                                socket, node.node) {
@@ -440,7 +444,7 @@ static int init_kernel_tracing(struct ltt_kernel_session *session)
                        assert(socket->fd >= 0);
 
                        pthread_mutex_lock(socket->lock);
-                       ret = kernel_consumer_send_session(socket->fd, session);
+                       ret = kernel_consumer_send_session(socket, session);
                        pthread_mutex_unlock(socket->lock);
                        if (ret < 0) {
                                ret = LTTNG_ERR_KERN_CONSUMER_FAIL;
@@ -450,6 +454,7 @@ static int init_kernel_tracing(struct ltt_kernel_session *session)
        }
 
 error:
+       rcu_read_unlock();
        return ret;
 }
 
@@ -528,7 +533,7 @@ error:
  */
 static int send_consumer_relayd_socket(int domain, struct ltt_session *session,
                struct lttng_uri *relayd_uri, struct consumer_output *consumer,
-               int consumer_fd)
+               struct consumer_socket *consumer_sock)
 {
        int ret;
        struct lttcomm_sock *sock = NULL;
@@ -557,8 +562,8 @@ static int send_consumer_relayd_socket(int domain, struct ltt_session *session,
        }
 
        /* Send relayd socket to consumer. */
-       ret = consumer_send_relayd_socket(consumer_fd, sock,
-                       consumer, relayd_uri->stype);
+       ret = consumer_send_relayd_socket(consumer_sock, sock,
+                       consumer, relayd_uri->stype, session->id);
        if (ret < 0) {
                ret = LTTNG_ERR_ENABLE_CONSUMER_FAIL;
                goto close_sock;
@@ -566,9 +571,9 @@ static int send_consumer_relayd_socket(int domain, struct ltt_session *session,
 
        /* Flag that the corresponding socket was sent. */
        if (relayd_uri->stype == LTTNG_STREAM_CONTROL) {
-               consumer->dst.net.control_sock_sent = 1;
+               consumer_sock->control_sock_sent = 1;
        } else if (relayd_uri->stype == LTTNG_STREAM_DATA) {
-               consumer->dst.net.data_sock_sent = 1;
+               consumer_sock->data_sock_sent = 1;
        }
 
        ret = LTTNG_OK;
@@ -593,7 +598,8 @@ close_sock:
  * session.
  */
 static int send_consumer_relayd_sockets(int domain,
-               struct ltt_session *session, struct consumer_output *consumer, int fd)
+               struct ltt_session *session, struct consumer_output *consumer,
+               struct consumer_socket *sock)
 {
        int ret = LTTNG_OK;
 
@@ -601,18 +607,18 @@ static int send_consumer_relayd_sockets(int domain,
        assert(consumer);
 
        /* Sending control relayd socket. */
-       if (!consumer->dst.net.control_sock_sent) {
+       if (!sock->control_sock_sent) {
                ret = send_consumer_relayd_socket(domain, session,
-                               &consumer->dst.net.control, consumer, fd);
+                               &consumer->dst.net.control, consumer, sock);
                if (ret != LTTNG_OK) {
                        goto error;
                }
        }
 
        /* Sending data relayd socket. */
-       if (!consumer->dst.net.data_sock_sent) {
+       if (!sock->data_sock_sent) {
                ret = send_consumer_relayd_socket(domain, session,
-                               &consumer->dst.net.data, consumer, fd);
+                               &consumer->dst.net.data, consumer, sock);
                if (ret != LTTNG_OK) {
                        goto error;
                }
@@ -642,6 +648,8 @@ static int setup_relayd(struct ltt_session *session)
 
        DBG2("Setting relayd for session %s", session->name);
 
+       rcu_read_lock();
+
        if (usess && usess->consumer && usess->consumer->type == CONSUMER_DST_NET
                        && usess->consumer->enabled) {
                /* For each consumer socket, send relayd sockets */
@@ -652,7 +660,7 @@ static int setup_relayd(struct ltt_session *session)
 
                        pthread_mutex_lock(socket->lock);
                        ret = send_consumer_relayd_sockets(LTTNG_DOMAIN_UST, session,
-                                       usess->consumer, socket->fd);
+                                       usess->consumer, socket);
                        pthread_mutex_unlock(socket->lock);
                        if (ret != LTTNG_OK) {
                                goto error;
@@ -669,7 +677,7 @@ static int setup_relayd(struct ltt_session *session)
 
                        pthread_mutex_lock(socket->lock);
                        ret = send_consumer_relayd_sockets(LTTNG_DOMAIN_KERNEL, session,
-                                       ksess->consumer, socket->fd);
+                                       ksess->consumer, socket);
                        pthread_mutex_unlock(socket->lock);
                        if (ret != LTTNG_OK) {
                                goto error;
@@ -678,6 +686,7 @@ static int setup_relayd(struct ltt_session *session)
        }
 
 error:
+       rcu_read_unlock();
        return ret;
 }
 
@@ -757,6 +766,8 @@ int cmd_disable_channel(struct ltt_session *session, int domain,
 
        usess = session->ust_session;
 
+       rcu_read_lock();
+
        switch (domain) {
        case LTTNG_DOMAIN_KERNEL:
        {
@@ -801,6 +812,7 @@ int cmd_disable_channel(struct ltt_session *session, int domain,
        ret = LTTNG_OK;
 
 error:
+       rcu_read_unlock();
        return ret;
 }
 
@@ -821,6 +833,8 @@ int cmd_enable_channel(struct ltt_session *session,
 
        DBG("Enabling channel %s for session %s", attr->name, session->name);
 
+       rcu_read_lock();
+
        switch (domain) {
        case LTTNG_DOMAIN_KERNEL:
        {
@@ -889,6 +903,7 @@ int cmd_enable_channel(struct ltt_session *session,
        }
 
 error:
+       rcu_read_unlock();
        return ret;
 }
 
@@ -901,6 +916,8 @@ int cmd_disable_event(struct ltt_session *session, int domain,
 {
        int ret;
 
+       rcu_read_lock();
+
        switch (domain) {
        case LTTNG_DOMAIN_KERNEL:
        {
@@ -959,6 +976,7 @@ int cmd_disable_event(struct ltt_session *session, int domain,
        ret = LTTNG_OK;
 
 error:
+       rcu_read_unlock();
        return ret;
 }
 
@@ -970,6 +988,8 @@ int cmd_disable_event_all(struct ltt_session *session, int domain,
 {
        int ret;
 
+       rcu_read_lock();
+
        switch (domain) {
        case LTTNG_DOMAIN_KERNEL:
        {
@@ -1028,6 +1048,7 @@ int cmd_disable_event_all(struct ltt_session *session, int domain,
        ret = LTTNG_OK;
 
 error:
+       rcu_read_unlock();
        return ret;
 }
 
@@ -1117,6 +1138,8 @@ int cmd_enable_event(struct ltt_session *session, int domain,
        assert(event);
        assert(channel_name);
 
+       rcu_read_lock();
+
        switch (domain) {
        case LTTNG_DOMAIN_KERNEL:
        {
@@ -1210,6 +1233,7 @@ int cmd_enable_event(struct ltt_session *session, int domain,
        ret = LTTNG_OK;
 
 error:
+       rcu_read_unlock();
        return ret;
 }
 
@@ -1226,6 +1250,8 @@ int cmd_enable_event_all(struct ltt_session *session, int domain,
        assert(session);
        assert(channel_name);
 
+       rcu_read_lock();
+
        switch (domain) {
        case LTTNG_DOMAIN_KERNEL:
        {
@@ -1356,6 +1382,7 @@ int cmd_enable_event_all(struct ltt_session *session, int domain,
        ret = LTTNG_OK;
 
 error:
+       rcu_read_unlock();
        return ret;
 }
 
@@ -1635,15 +1662,8 @@ int cmd_set_consumer_uri(int domain, struct ltt_session *session,
                        goto error;
                }
 
-               /*
-                * Don't send relayd socket if URI is NOT remote or if the relayd
-                * socket for the session was already sent.
-                */
-               if (uris[i].dtype == LTTNG_DST_PATH ||
-                               (uris[i].stype == LTTNG_STREAM_CONTROL &&
-                               consumer->dst.net.control_sock_sent) ||
-                               (uris[i].stype == LTTNG_STREAM_DATA &&
-                               consumer->dst.net.data_sock_sent)) {
+               /* Don't send relayd socket if URI is NOT remote */
+               if (uris[i].dtype == LTTNG_DST_PATH) {
                        continue;
                }
 
@@ -1657,7 +1677,7 @@ int cmd_set_consumer_uri(int domain, struct ltt_session *session,
 
                        pthread_mutex_lock(socket->lock);
                        ret = send_consumer_relayd_socket(domain, session, &uris[i],
-                                       consumer, socket->fd);
+                                       consumer, socket);
                        pthread_mutex_unlock(socket->lock);
                        if (ret != LTTNG_OK) {
                                rcu_read_unlock();