Fix: hold consumer socket lock for consumer_send_msg

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

diff --git a/src/bin/lttng-sessiond/ust-consumer.c b/src/bin/lttng-sessiond/ust-consumer.c
index 27ebf7d23270a6ad4a5a1e13f3eb971573a2455a..b9865a0f015b76e0abb5ebcedfd8b92238b47465 100644 (file)
--- a/src/bin/lttng-sessiond/ust-consumer.c
+++ b/src/bin/lttng-sessiond/ust-consumer.c
@@ -15,7 +15,6 @@
  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  */
 
-#define _GNU_SOURCE
 #define _LGPL_SOURCE
 #include <errno.h>
 #include <stdio.h>
@@ -25,7 +24,7 @@
 #include <inttypes.h>
 
 #include <common/common.h>
-#include <common/consumer.h>
+#include <common/consumer/consumer.h>
 #include <common/defaults.h>
 
 #include "consumer.h"
@@ -33,6 +32,7 @@
 #include "ust-consumer.h"
 #include "buffer-registry.h"
 #include "session.h"
+#include "lttng-sessiond.h"
 
 /*
  * Return allocated full pathname of the session using the consumer trace path
@@ -53,8 +53,11 @@ static char *setup_trace_path(struct consumer_output *consumer,
 
        health_code_update();
 
-       /* Allocate our self the string to make sure we never exceed PATH_MAX. */
-       pathname = zmalloc(PATH_MAX);
+       /*
+        * Allocate the string ourself to make sure we never exceed
+        * LTTNG_PATH_MAX.
+        */
+       pathname = zmalloc(LTTNG_PATH_MAX);
        if (!pathname) {
                goto error;
        }
@@ -62,8 +65,10 @@ static char *setup_trace_path(struct consumer_output *consumer,
        /* Get correct path name destination */
        if (consumer->type == CONSUMER_DST_LOCAL) {
                /* Set application path to the destination path */
-               ret = snprintf(pathname, PATH_MAX, "%s%s%s",
-                               consumer->dst.trace_path, consumer->subdir, ua_sess->path);
+               ret = snprintf(pathname, LTTNG_PATH_MAX, "%s%s%s%s",
+                               consumer->dst.session_root_path,
+                               consumer->chunk_path,
+                               consumer->subdir, ua_sess->path);
                if (ret < 0) {
                        PERROR("snprintf channel path");
                        goto error;
@@ -79,7 +84,10 @@ static char *setup_trace_path(struct consumer_output *consumer,
                        }
                }
        } else {
-               ret = snprintf(pathname, PATH_MAX, "%s%s", consumer->subdir,
+               ret = snprintf(pathname, LTTNG_PATH_MAX, "%s%s%s%s",
+                               consumer->dst.net.base_dir,
+                               consumer->chunk_path,
+                               consumer->subdir,
                                ua_sess->path);
                if (ret < 0) {
                        PERROR("snprintf channel path");
@@ -95,7 +103,7 @@ error:
 }
 
 /*
- * Send a single channel to the consumer using command ADD_CHANNEL.
+ * Send a single channel to the consumer using command ASK_CHANNEL_CREATION.
  *
  * Consumer socket lock MUST be acquired before calling this.
  */
@@ -175,6 +183,7 @@ static int ask_channel_creation(struct ust_app_session *ua_sess,
                        ua_chan->attr.switch_timer_interval,
                        ua_chan->attr.read_timer_interval,
                        ua_sess->live_timer_interval,
+                       ua_chan->monitor_timer_interval,
                        output,
                        (int) ua_chan->attr.type,
                        ua_sess->tracing_id,
@@ -191,6 +200,7 @@ static int ask_channel_creation(struct ust_app_session *ua_sess,
                        ua_sess->id,
                        ua_sess->output_traces,
                        ua_sess->uid,
+                       ua_chan->attr.blocking_timeout,
                        root_shm_path, shm_path);
 
        health_code_update();
@@ -224,6 +234,8 @@ error:
 /*
  * Ask consumer to create a channel for a given session.
  *
+ * Session list and rcu read side locks must be held by the caller.
+ *
  * Returns 0 on success else a negative value.
  */
 int ust_consumer_ask_channel(struct ust_app_session *ua_sess,
@@ -231,6 +243,7 @@ int ust_consumer_ask_channel(struct ust_app_session *ua_sess,
                struct consumer_socket *socket, struct ust_registry_session *registry)
 {
        int ret;
+       struct ltt_session *session;
 
        assert(ua_sess);
        assert(ua_chan);
@@ -244,10 +257,14 @@ int ust_consumer_ask_channel(struct ust_app_session *ua_sess,
                goto error;
        }
 
+       session = session_find_by_id(ua_sess->tracing_id);
+       assert(session);
+
        pthread_mutex_lock(socket->lock);
        ret = ask_channel_creation(ua_sess, ua_chan, consumer, socket, registry);
        pthread_mutex_unlock(socket->lock);
        if (ret < 0) {
+               ERR("ask_channel_creation consumer command failed");
                goto error;
        }
 
@@ -312,7 +329,6 @@ int ust_consumer_get_channel(struct consumer_socket *socket,
                        free(stream);
                        if (ret == -LTTNG_UST_ERR_NOENT) {
                                DBG3("UST app consumer has no more stream available");
-                               ret = 0;
                                break;
                        }
                        if (ret != -EPIPE) {
@@ -400,7 +416,9 @@ int ust_consumer_send_stream_to_ust(struct ust_app *app,
        DBG2("UST consumer send stream to app %d", app->sock);
 
        /* Relay stream to application. */
+       pthread_mutex_lock(&app->sock_lock);
        ret = ustctl_send_stream_to_ust(app->sock, channel->obj, stream->obj);
+       pthread_mutex_unlock(&app->sock_lock);
        if (ret < 0) {
                if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
                        ERR("ustctl send stream handle %d to app pid: %d with ret %d",
@@ -435,7 +453,9 @@ int ust_consumer_send_channel_to_ust(struct ust_app *app,
                        app->sock, app->pid, channel->name, channel->tracing_channel_id);
 
        /* Send stream to application. */
+       pthread_mutex_lock(&app->sock_lock);
        ret = ustctl_send_channel_to_ust(app->sock, ua_sess->handle, channel->obj);
+       pthread_mutex_unlock(&app->sock_lock);
        if (ret < 0) {
                if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
                        ERR("Error ustctl send channel %s to app pid: %d with ret %d",
@@ -493,7 +513,9 @@ int ust_consumer_metadata_request(struct consumer_socket *socket)
 
                        memset(&msg, 0, sizeof(msg));
                        msg.cmd_type = LTTNG_ERR_UND;
+                       pthread_mutex_lock(socket->lock);
                        (void) consumer_send_msg(socket, &msg);
+                       pthread_mutex_unlock(socket->lock);
                        /*
                         * This is possible since the session might have been destroyed
                         * during a consumer metadata request. So here, return gracefully