Move LTTng-UST buffer ownership from application to consumer
[lttng-tools.git] / src / bin / lttng-sessiond / cmd.c
index 06bf0f4f386832a199a199862e2416301d42e521..150a2dfe4675bfbfaf90d5e30b51d54ab6eb284d 100644 (file)
@@ -101,11 +101,22 @@ static int build_network_session_path(char *dst, size_t size,
                goto error;
        }
 
-       if (strlen(tmp_uurl) > 0) {
+       /*
+        * Do we have a UST url set. If yes, this means we have both kernel and UST
+        * to print.
+        */
+       if (*tmp_uurl != '\0') {
                ret = snprintf(dst, size, "[K]: %s [data: %d] -- [U]: %s [data: %d]",
                                tmp_urls, kdata_port, tmp_uurl, udata_port);
        } else {
-               ret = snprintf(dst, size, "%s [data: %d]", tmp_urls, kdata_port);
+               int dport;
+               if (kuri) {
+                       dport = kdata_port;
+               } else {
+                       /* No kernel URI, use the UST port. */
+                       dport = udata_port;
+               }
+               ret = snprintf(dst, size, "%s [data: %d]", tmp_urls, dport);
        }
 
 error:
@@ -141,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);
@@ -160,6 +172,7 @@ static void list_lttng_channels(int domain, struct ltt_session *session,
                        }
                        i++;
                }
+               rcu_read_unlock();
                break;
        }
        default:
@@ -363,6 +376,8 @@ static int add_uri_to_consumer(struct consumer_output *consumer,
        case LTTNG_DST_IPV6:
                DBG2("Setting network URI to consumer");
 
+               consumer->type = CONSUMER_DST_NET;
+
                /* Set URI into consumer output object */
                ret = consumer_set_network_uri(consumer, uri);
                if (ret < 0) {
@@ -422,6 +437,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) {
@@ -429,7 +446,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;
@@ -439,6 +456,7 @@ static int init_kernel_tracing(struct ltt_kernel_session *session)
        }
 
 error:
+       rcu_read_unlock();
        return ret;
 }
 
@@ -468,8 +486,14 @@ static int create_connect_relayd(struct consumer_output *output,
                goto error;
        }
 
-       /* Connect to relayd so we can proceed with a session creation. */
+       /*
+        * Connect to relayd so we can proceed with a session creation. This call
+        * can possibly block for an arbitrary amount of time to set the health
+        * state to be in poll execution.
+        */
+       health_poll_entry();
        ret = relayd_connect(sock);
+       health_poll_exit();
        if (ret < 0) {
                ERR("Unable to reach lttng-relayd");
                ret = LTTNG_ERR_RELAYD_CONNECT_FAIL;
@@ -517,11 +541,22 @@ 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;
 
+       /* Connect to relayd and make version check if uri is the control. */
+       ret = create_connect_relayd(consumer, session->name, relayd_uri, &sock);
+       if (ret != LTTNG_OK) {
+               goto close_sock;
+       }
+
+       /* If the control socket is connected, network session is ready */
+       if (relayd_uri->stype == LTTNG_STREAM_CONTROL) {
+               session->net_handle = 1;
+       }
+
        /* Set the network sequence index if not set. */
        if (consumer->net_seq_index == -1) {
                /*
@@ -534,25 +569,21 @@ static int send_consumer_relayd_socket(int domain, struct ltt_session *session,
                                uatomic_read(&relayd_net_seq_idx));
        }
 
-       /* Connect to relayd and make version check if uri is the control. */
-       ret = create_connect_relayd(consumer, session->name, relayd_uri, &sock);
-       if (ret != LTTNG_OK) {
-               goto close_sock;
-       }
-
-       /* If the control socket is connected, network session is ready */
-       if (relayd_uri->stype == LTTNG_STREAM_CONTROL) {
-               session->net_handle = 1;
-       }
-
        /* 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;
        }
 
+       /* Flag that the corresponding socket was sent. */
+       if (relayd_uri->stype == LTTNG_STREAM_CONTROL) {
+               consumer_sock->control_sock_sent = 1;
+       } else if (relayd_uri->stype == LTTNG_STREAM_DATA) {
+               consumer_sock->data_sock_sent = 1;
+       }
+
        ret = LTTNG_OK;
 
        /*
@@ -566,6 +597,14 @@ close_sock:
                lttcomm_destroy_sock(sock);
        }
 
+       if (ret != LTTNG_OK) {
+               /*
+                * On error, nullify the consumer sequence index so streams are not
+                * associated with it once sent to the consumer.
+                */
+               uatomic_set(&consumer->net_seq_index, -1);
+       }
+
        return ret;
 }
 
@@ -575,36 +614,32 @@ 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;
+       int ret = LTTNG_OK;
 
        assert(session);
        assert(consumer);
 
-       /* Don't resend the sockets to the consumer. */
-       if (consumer->dst.net.relayd_socks_sent) {
-               ret = LTTNG_OK;
-               goto error;
-       }
-
        /* Sending control relayd socket. */
-       ret = send_consumer_relayd_socket(domain, session,
-                       &consumer->dst.net.control, consumer, fd);
-       if (ret != LTTNG_OK) {
-               goto error;
+       if (!sock->control_sock_sent) {
+               ret = send_consumer_relayd_socket(domain, session,
+                               &consumer->dst.net.control, consumer, sock);
+               if (ret != LTTNG_OK) {
+                       goto error;
+               }
        }
 
        /* Sending data relayd socket. */
-       ret = send_consumer_relayd_socket(domain, session,
-                       &consumer->dst.net.data, consumer, fd);
-       if (ret != LTTNG_OK) {
-               goto error;
+       if (!sock->data_sock_sent) {
+               ret = send_consumer_relayd_socket(domain, session,
+                               &consumer->dst.net.data, consumer, sock);
+               if (ret != LTTNG_OK) {
+                       goto error;
+               }
        }
 
-       /* Flag that all relayd sockets were sent to the consumer. */
-       consumer->dst.net.relayd_socks_sent = 1;
-
 error:
        return ret;
 }
@@ -614,7 +649,7 @@ error:
  * the relayd and send them to the right domain consumer. Consumer type MUST be
  * network.
  */
-static int setup_relayd(struct ltt_session *session)
+int cmd_setup_relayd(struct ltt_session *session)
 {
        int ret = LTTNG_OK;
        struct ltt_ust_session *usess;
@@ -629,6 +664,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 */
@@ -638,8 +675,8 @@ static int setup_relayd(struct ltt_session *session)
                        assert(socket->fd >= 0);
 
                        pthread_mutex_lock(socket->lock);
-                       send_consumer_relayd_sockets(LTTNG_DOMAIN_UST, session,
-                                       usess->consumer, socket->fd);
+                       ret = send_consumer_relayd_sockets(LTTNG_DOMAIN_UST, session,
+                                       usess->consumer, socket);
                        pthread_mutex_unlock(socket->lock);
                        if (ret != LTTNG_OK) {
                                goto error;
@@ -655,8 +692,8 @@ static int setup_relayd(struct ltt_session *session)
                        assert(socket->fd >= 0);
 
                        pthread_mutex_lock(socket->lock);
-                       send_consumer_relayd_sockets(LTTNG_DOMAIN_KERNEL, session,
-                                       ksess->consumer, socket->fd);
+                       ret = send_consumer_relayd_sockets(LTTNG_DOMAIN_KERNEL, session,
+                                       ksess->consumer, socket);
                        pthread_mutex_unlock(socket->lock);
                        if (ret != LTTNG_OK) {
                                goto error;
@@ -664,6 +701,72 @@ static int setup_relayd(struct ltt_session *session)
                }
        }
 
+error:
+       rcu_read_unlock();
+       return ret;
+}
+
+/*
+ * Start a kernel session by opening all necessary streams.
+ */
+static int start_kernel_session(struct ltt_kernel_session *ksess, int wpipe)
+{
+       int ret;
+       struct ltt_kernel_channel *kchan;
+
+       /* Open kernel metadata */
+       if (ksess->metadata == NULL) {
+               ret = kernel_open_metadata(ksess);
+               if (ret < 0) {
+                       ret = LTTNG_ERR_KERN_META_FAIL;
+                       goto error;
+               }
+       }
+
+       /* Open kernel metadata stream */
+       if (ksess->metadata_stream_fd < 0) {
+               ret = kernel_open_metadata_stream(ksess);
+               if (ret < 0) {
+                       ERR("Kernel create metadata stream failed");
+                       ret = LTTNG_ERR_KERN_STREAM_FAIL;
+                       goto error;
+               }
+       }
+
+       /* For each channel */
+       cds_list_for_each_entry(kchan, &ksess->channel_list.head, list) {
+               if (kchan->stream_count == 0) {
+                       ret = kernel_open_channel_stream(kchan);
+                       if (ret < 0) {
+                               ret = LTTNG_ERR_KERN_STREAM_FAIL;
+                               goto error;
+                       }
+                       /* Update the stream global counter */
+                       ksess->stream_count_global += ret;
+               }
+       }
+
+       /* Setup kernel consumer socket and send fds to it */
+       ret = init_kernel_tracing(ksess);
+       if (ret < 0) {
+               ret = LTTNG_ERR_KERN_START_FAIL;
+               goto error;
+       }
+
+       /* This start the kernel tracing */
+       ret = kernel_start_session(ksess);
+       if (ret < 0) {
+               ret = LTTNG_ERR_KERN_START_FAIL;
+               goto error;
+       }
+
+       /* Quiescent wait after starting trace */
+       kernel_wait_quiescent(kernel_tracer_fd);
+
+       ksess->started = 1;
+
+       ret = LTTNG_OK;
+
 error:
        return ret;
 }
@@ -679,6 +782,8 @@ int cmd_disable_channel(struct ltt_session *session, int domain,
 
        usess = session->ust_session;
 
+       rcu_read_lock();
+
        switch (domain) {
        case LTTNG_DOMAIN_KERNEL:
        {
@@ -723,6 +828,7 @@ int cmd_disable_channel(struct ltt_session *session, int domain,
        ret = LTTNG_OK;
 
 error:
+       rcu_read_unlock();
        return ret;
 }
 
@@ -743,14 +849,13 @@ 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:
        {
                struct ltt_kernel_channel *kchan;
 
-               /* Mandatory for a kernel channel. */
-               assert(wpipe > 0);
-
                kchan = trace_kernel_get_channel_by_name(attr->name,
                                session->kernel_session);
                if (kchan == NULL) {
@@ -764,6 +869,18 @@ int cmd_enable_channel(struct ltt_session *session,
                }
 
                kernel_wait_quiescent(kernel_tracer_fd);
+
+               /*
+                * If the session was previously started, start as well this newly
+                * created kernel session so the events/channels enabled *after* the
+                * start actually work.
+                */
+               if (session->started && !session->kernel_session->started) {
+                       ret = start_kernel_session(session->kernel_session, wpipe);
+                       if (ret != LTTNG_OK) {
+                               goto error;
+                       }
+               }
                break;
        }
        case LTTNG_DOMAIN_UST:
@@ -778,6 +895,17 @@ int cmd_enable_channel(struct ltt_session *session,
                } else {
                        ret = channel_ust_enable(usess, domain, uchan);
                }
+
+               /* Start the UST session if the session was already started. */
+               if (session->started && !usess->start_trace) {
+                       ret = ust_app_start_trace_all(usess);
+                       if (ret < 0) {
+                               ret = LTTNG_ERR_UST_START_FAIL;
+                               goto error;
+                       }
+                       ret = LTTNG_OK;
+                       usess->start_trace = 1;
+               }
                break;
        }
 #if 0
@@ -791,6 +919,7 @@ int cmd_enable_channel(struct ltt_session *session,
        }
 
 error:
+       rcu_read_unlock();
        return ret;
 }
 
@@ -803,6 +932,8 @@ int cmd_disable_event(struct ltt_session *session, int domain,
 {
        int ret;
 
+       rcu_read_lock();
+
        switch (domain) {
        case LTTNG_DOMAIN_KERNEL:
        {
@@ -817,7 +948,7 @@ int cmd_disable_event(struct ltt_session *session, int domain,
                        goto error;
                }
 
-               ret = event_kernel_disable_tracepoint(ksess, kchan, event_name);
+               ret = event_kernel_disable_tracepoint(kchan, event_name);
                if (ret != LTTNG_OK) {
                        goto error;
                }
@@ -861,6 +992,7 @@ int cmd_disable_event(struct ltt_session *session, int domain,
        ret = LTTNG_OK;
 
 error:
+       rcu_read_unlock();
        return ret;
 }
 
@@ -872,6 +1004,8 @@ int cmd_disable_event_all(struct ltt_session *session, int domain,
 {
        int ret;
 
+       rcu_read_lock();
+
        switch (domain) {
        case LTTNG_DOMAIN_KERNEL:
        {
@@ -886,7 +1020,7 @@ int cmd_disable_event_all(struct ltt_session *session, int domain,
                        goto error;
                }
 
-               ret = event_kernel_disable_all(ksess, kchan);
+               ret = event_kernel_disable_all(kchan);
                if (ret != LTTNG_OK) {
                        goto error;
                }
@@ -930,6 +1064,7 @@ int cmd_disable_event_all(struct ltt_session *session, int domain,
        ret = LTTNG_OK;
 
 error:
+       rcu_read_unlock();
        return ret;
 }
 
@@ -937,15 +1072,24 @@ error:
  * Command LTTNG_ADD_CONTEXT processed by the client thread.
  */
 int cmd_add_context(struct ltt_session *session, int domain,
-               char *channel_name, char *event_name, struct lttng_event_context *ctx)
+               char *channel_name, struct lttng_event_context *ctx, int kwpipe)
 {
        int ret;
 
        switch (domain) {
        case LTTNG_DOMAIN_KERNEL:
+               assert(session->kernel_session);
+
+               if (session->kernel_session->channel_count == 0) {
+                       /* Create default channel */
+                       ret = channel_kernel_create(session->kernel_session, NULL, kwpipe);
+                       if (ret != LTTNG_OK) {
+                               goto error;
+                       }
+               }
+
                /* Add kernel context to kernel tracer */
-               ret = context_kernel_add(session->kernel_session, ctx,
-                               event_name, channel_name);
+               ret = context_kernel_add(session->kernel_session, ctx, channel_name);
                if (ret != LTTNG_OK) {
                        goto error;
                }
@@ -953,49 +1097,28 @@ int cmd_add_context(struct ltt_session *session, int domain,
        case LTTNG_DOMAIN_UST:
        {
                struct ltt_ust_session *usess = session->ust_session;
-
                assert(usess);
 
-               ret = context_ust_add(usess, domain, ctx, event_name, channel_name);
-               if (ret != LTTNG_OK) {
-                       goto error;
-               }
-               break;
-       }
-#if 0
-       case LTTNG_DOMAIN_UST_EXEC_NAME:
-       case LTTNG_DOMAIN_UST_PID:
-       case LTTNG_DOMAIN_UST_PID_FOLLOW_CHILDREN:
-#endif
-       default:
-               ret = LTTNG_ERR_UND;
-               goto error;
-       }
-
-       ret = LTTNG_OK;
-
-error:
-       return ret;
-}
-
-/*
- * Command LTTNG_SET_FILTER processed by the client thread.
- */
-int cmd_set_filter(struct ltt_session *session, int domain,
-               char *channel_name, char *event_name,
-               struct lttng_filter_bytecode *bytecode)
-{
-       int ret;
+               unsigned int chan_count =
+                       lttng_ht_get_count(usess->domain_global.channels);
+               if (chan_count == 0) {
+                       struct lttng_channel *attr;
+                       /* Create default channel */
+                       attr = channel_new_default_attr(domain);
+                       if (attr == NULL) {
+                               ret = LTTNG_ERR_FATAL;
+                               goto error;
+                       }
 
-       switch (domain) {
-       case LTTNG_DOMAIN_KERNEL:
-               ret = LTTNG_ERR_FATAL;
-               break;
-       case LTTNG_DOMAIN_UST:
-       {
-               struct ltt_ust_session *usess = session->ust_session;
+                       ret = channel_ust_create(usess, domain, attr);
+                       if (ret != LTTNG_OK) {
+                               free(attr);
+                               goto error;
+                       }
+                       free(attr);
+               }
 
-               ret = filter_ust_set(usess, domain, bytecode, event_name, channel_name);
+               ret = context_ust_add(usess, domain, ctx, channel_name);
                if (ret != LTTNG_OK) {
                        goto error;
                }
@@ -1015,23 +1138,24 @@ int cmd_set_filter(struct ltt_session *session, int domain,
 
 error:
        return ret;
-
 }
 
-
 /*
  * Command LTTNG_ENABLE_EVENT processed by the client thread.
  */
 int cmd_enable_event(struct ltt_session *session, int domain,
-               char *channel_name, struct lttng_event *event, int wpipe)
+               char *channel_name, struct lttng_event *event,
+               struct lttng_filter_bytecode *filter, int wpipe)
 {
-       int ret;
+       int ret, channel_created = 0;
        struct lttng_channel *attr;
 
        assert(session);
        assert(event);
        assert(channel_name);
 
+       rcu_read_lock();
+
        switch (domain) {
        case LTTNG_DOMAIN_KERNEL:
        {
@@ -1053,6 +1177,8 @@ int cmd_enable_event(struct ltt_session *session, int domain,
                                goto error;
                        }
                        free(attr);
+
+                       channel_created = 1;
                }
 
                /* Get the newly created kernel channel pointer */
@@ -1064,9 +1190,12 @@ int cmd_enable_event(struct ltt_session *session, int domain,
                        goto error;
                }
 
-               ret = event_kernel_enable_tracepoint(session->kernel_session, kchan,
-                               event);
+               ret = event_kernel_enable_tracepoint(kchan, event);
                if (ret != LTTNG_OK) {
+                       if (channel_created) {
+                               /* Let's not leak a useless channel. */
+                               kernel_destroy_channel(kchan);
+                       }
                        goto error;
                }
 
@@ -1106,7 +1235,7 @@ int cmd_enable_event(struct ltt_session *session, int domain,
                }
 
                /* At this point, the session and channel exist on the tracer */
-               ret = event_ust_enable_tracepoint(usess, domain, uchan, event);
+               ret = event_ust_enable_tracepoint(usess, domain, uchan, event, filter);
                if (ret != LTTNG_OK) {
                        goto error;
                }
@@ -1125,6 +1254,7 @@ int cmd_enable_event(struct ltt_session *session, int domain,
        ret = LTTNG_OK;
 
 error:
+       rcu_read_unlock();
        return ret;
 }
 
@@ -1132,7 +1262,8 @@ error:
  * Command LTTNG_ENABLE_ALL_EVENT processed by the client thread.
  */
 int cmd_enable_event_all(struct ltt_session *session, int domain,
-               char *channel_name, int event_type, int wpipe)
+               char *channel_name, int event_type,
+               struct lttng_filter_bytecode *filter, int wpipe)
 {
        int ret;
        struct lttng_channel *attr;
@@ -1140,6 +1271,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:
        {
@@ -1173,21 +1306,18 @@ int cmd_enable_event_all(struct ltt_session *session, int domain,
 
                switch (event_type) {
                case LTTNG_EVENT_SYSCALL:
-                       ret = event_kernel_enable_all_syscalls(session->kernel_session,
-                                       kchan, kernel_tracer_fd);
+                       ret = event_kernel_enable_all_syscalls(kchan, kernel_tracer_fd);
                        break;
                case LTTNG_EVENT_TRACEPOINT:
                        /*
                         * This call enables all LTTNG_KERNEL_TRACEPOINTS and
                         * events already registered to the channel.
                         */
-                       ret = event_kernel_enable_all_tracepoints(session->kernel_session,
-                                       kchan, kernel_tracer_fd);
+                       ret = event_kernel_enable_all_tracepoints(kchan, kernel_tracer_fd);
                        break;
                case LTTNG_EVENT_ALL:
                        /* Enable syscalls and tracepoints */
-                       ret = event_kernel_enable_all(session->kernel_session,
-                                       kchan, kernel_tracer_fd);
+                       ret = event_kernel_enable_all(kchan, kernel_tracer_fd);
                        break;
                default:
                        ret = LTTNG_ERR_KERN_ENABLE_FAIL;
@@ -1196,6 +1326,10 @@ int cmd_enable_event_all(struct ltt_session *session, int domain,
 
                /* Manage return value */
                if (ret != LTTNG_OK) {
+                       /*
+                        * On error, cmd_enable_channel call will take care of destroying
+                        * the created channel if it was needed.
+                        */
                        goto error;
                }
 
@@ -1239,7 +1373,8 @@ int cmd_enable_event_all(struct ltt_session *session, int domain,
                switch (event_type) {
                case LTTNG_EVENT_ALL:
                case LTTNG_EVENT_TRACEPOINT:
-                       ret = event_ust_enable_all_tracepoints(usess, domain, uchan);
+                       ret = event_ust_enable_all_tracepoints(usess, domain, uchan,
+                                       filter);
                        if (ret != LTTNG_OK) {
                                goto error;
                        }
@@ -1269,6 +1404,7 @@ int cmd_enable_event_all(struct ltt_session *session, int domain,
        ret = LTTNG_OK;
 
 error:
+       rcu_read_unlock();
        return ret;
 }
 
@@ -1346,7 +1482,6 @@ int cmd_start_trace(struct ltt_session *session)
        int ret;
        struct ltt_kernel_session *ksession;
        struct ltt_ust_session *usess;
-       struct ltt_kernel_channel *kchan;
 
        assert(session);
 
@@ -1362,62 +1497,12 @@ int cmd_start_trace(struct ltt_session *session)
 
        session->enabled = 1;
 
-       ret = setup_relayd(session);
-       if (ret != LTTNG_OK) {
-               ERR("Error setting up relayd for session %s", session->name);
-               goto error;
-       }
-
        /* Kernel tracing */
        if (ksession != NULL) {
-               /* Open kernel metadata */
-               if (ksession->metadata == NULL) {
-                       ret = kernel_open_metadata(ksession);
-                       if (ret < 0) {
-                               ret = LTTNG_ERR_KERN_META_FAIL;
-                               goto error;
-                       }
-               }
-
-               /* Open kernel metadata stream */
-               if (ksession->metadata_stream_fd < 0) {
-                       ret = kernel_open_metadata_stream(ksession);
-                       if (ret < 0) {
-                               ERR("Kernel create metadata stream failed");
-                               ret = LTTNG_ERR_KERN_STREAM_FAIL;
-                               goto error;
-                       }
-               }
-
-               /* For each channel */
-               cds_list_for_each_entry(kchan, &ksession->channel_list.head, list) {
-                       if (kchan->stream_count == 0) {
-                               ret = kernel_open_channel_stream(kchan);
-                               if (ret < 0) {
-                                       ret = LTTNG_ERR_KERN_STREAM_FAIL;
-                                       goto error;
-                               }
-                               /* Update the stream global counter */
-                               ksession->stream_count_global += ret;
-                       }
-               }
-
-               /* Setup kernel consumer socket and send fds to it */
-               ret = init_kernel_tracing(ksession);
-               if (ret < 0) {
-                       ret = LTTNG_ERR_KERN_START_FAIL;
-                       goto error;
-               }
-
-               /* This start the kernel tracing */
-               ret = kernel_start_session(ksession);
-               if (ret < 0) {
-                       ret = LTTNG_ERR_KERN_START_FAIL;
+               ret = start_kernel_session(ksession, kernel_tracer_fd);
+               if (ret != LTTNG_OK) {
                        goto error;
                }
-
-               /* Quiescent wait after starting trace */
-               kernel_wait_quiescent(kernel_tracer_fd);
        }
 
        /* Flag session that trace should start automatically */
@@ -1431,6 +1516,8 @@ int cmd_start_trace(struct ltt_session *session)
                }
        }
 
+       session->started = 1;
+
        ret = LTTNG_OK;
 
 error:
@@ -1487,6 +1574,8 @@ int cmd_stop_trace(struct ltt_session *session)
                }
 
                kernel_wait_quiescent(kernel_tracer_fd);
+
+               ksession->started = 0;
        }
 
        if (usess) {
@@ -1499,6 +1588,8 @@ int cmd_stop_trace(struct ltt_session *session)
                }
        }
 
+       session->started = 0;
+
        ret = LTTNG_OK;
 
 error:
@@ -1526,11 +1617,6 @@ int cmd_set_consumer_uri(int domain, struct ltt_session *session,
                goto error;
        }
 
-       if (!session->start_consumer) {
-               ret = LTTNG_ERR_NO_CONSUMER;
-               goto error;
-       }
-
        /*
         * This case switch makes sure the domain session has a temporary consumer
         * so the URL can be set.
@@ -1584,41 +1670,10 @@ int cmd_set_consumer_uri(int domain, struct ltt_session *session,
        }
 
        for (i = 0; i < nb_uri; i++) {
-               struct consumer_socket *socket;
-               struct lttng_ht_iter iter;
-
                ret = add_uri_to_consumer(consumer, &uris[i], domain, session->name);
                if (ret < 0) {
                        goto error;
                }
-
-               /*
-                * Don't send relayd socket if URI is NOT remote or if the relayd
-                * sockets for the session are already sent.
-                */
-               if (uris[i].dtype == LTTNG_DST_PATH ||
-                               consumer->dst.net.relayd_socks_sent) {
-                       continue;
-               }
-
-               /* Try to send relayd URI to the consumer if exist. */
-               rcu_read_lock();
-               cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter,
-                               socket, node.node) {
-
-                       /* A socket in the HT should never have a negative fd */
-                       assert(socket->fd >= 0);
-
-                       pthread_mutex_lock(socket->lock);
-                       ret = send_consumer_relayd_socket(domain, session, &uris[i],
-                                       consumer, socket->fd);
-                       pthread_mutex_unlock(socket->lock);
-                       if (ret != LTTNG_OK) {
-                               rcu_read_unlock();
-                               goto error;
-                       }
-               }
-               rcu_read_unlock();
        }
 
        /* All good! */
@@ -1834,8 +1889,11 @@ int cmd_register_consumer(struct ltt_session *session, int domain,
 
                socket = consumer_allocate_socket(sock);
                if (socket == NULL) {
+                       ret = close(sock);
+                       if (ret < 0) {
+                               PERROR("close register consumer");
+                       }
                        ret = LTTNG_ERR_FATAL;
-                       close(sock);
                        goto error;
                }
 
@@ -2137,11 +2195,6 @@ int cmd_enable_consumer(int domain, struct ltt_session *session)
                goto error;
        }
 
-       if (!session->start_consumer) {
-               ret = LTTNG_ERR_NO_CONSUMER;
-               goto error;
-       }
-
        switch (domain) {
        case 0:
                assert(session->consumer);
@@ -2208,11 +2261,6 @@ int cmd_enable_consumer(int domain, struct ltt_session *session)
                        break;
                }
 
-               /* Append default kernel trace dir to subdir */
-               strncat(ksess->consumer->subdir, DEFAULT_KERNEL_TRACE_DIR,
-                               sizeof(ksess->consumer->subdir) -
-                               strlen(ksess->consumer->subdir) - 1);
-
                /*
                 * @session-lock
                 * This is race free for now since the session lock is acquired before
@@ -2221,6 +2269,7 @@ int cmd_enable_consumer(int domain, struct ltt_session *session)
                 * is valid.
                 */
                rcu_read_lock();
+               /* Destroy current consumer. We are about to replace it */
                consumer_destroy_output(ksess->consumer);
                rcu_read_unlock();
                ksess->consumer = consumer;
@@ -2294,11 +2343,6 @@ int cmd_enable_consumer(int domain, struct ltt_session *session)
                        break;
                }
 
-               /* Append default kernel trace dir to subdir */
-               strncat(usess->consumer->subdir, DEFAULT_UST_TRACE_DIR,
-                               sizeof(usess->consumer->subdir) -
-                               strlen(usess->consumer->subdir) - 1);
-
                /*
                 * @session-lock
                 * This is race free for now since the session lock is acquired before
@@ -2307,6 +2351,7 @@ int cmd_enable_consumer(int domain, struct ltt_session *session)
                 * is valid.
                 */
                rcu_read_lock();
+               /* Destroy current consumer. We are about to replace it */
                consumer_destroy_output(usess->consumer);
                rcu_read_unlock();
                usess->consumer = consumer;
@@ -2315,6 +2360,8 @@ int cmd_enable_consumer(int domain, struct ltt_session *session)
                break;
        }
 
+       session->start_consumer = 1;
+
        /* Enable it */
        if (consumer) {
                consumer->enabled = 1;
@@ -2329,6 +2376,47 @@ error:
        return ret;
 }
 
+/*
+ * Command LTTNG_DATA_PENDING returning 0 if the data is NOT pending meaning
+ * ready for trace analysis (or anykind of reader) or else 1 for pending data.
+ */
+int cmd_data_pending(struct ltt_session *session)
+{
+       int ret;
+       struct ltt_kernel_session *ksess = session->kernel_session;
+       struct ltt_ust_session *usess = session->ust_session;
+
+       assert(session);
+
+       /* Session MUST be stopped to ask for data availability. */
+       if (session->enabled) {
+               ret = LTTNG_ERR_SESSION_STARTED;
+               goto error;
+       }
+
+       if (ksess && ksess->consumer) {
+               ret = consumer_is_data_pending(ksess->id, ksess->consumer);
+               if (ret == 1) {
+                       /* Data is still being extracted for the kernel. */
+                       goto error;
+               }
+       }
+
+       if (usess && usess->consumer) {
+               ret = consumer_is_data_pending(usess->id, usess->consumer);
+               if (ret == 1) {
+                       /* Data is still being extracted for the kernel. */
+                       goto error;
+               }
+       }
+
+       /* Data is ready to be read by a viewer */
+       ret = 0;
+
+error:
+       return ret;
+}
+
 /*
  * Init command subsystem.
  */
This page took 0.037513 seconds and 4 git commands to generate.