Move LTTng-UST buffer ownership from application to consumer
authorDavid Goulet <dgoulet@efficios.com>
Wed, 30 Jan 2013 21:36:23 +0000 (16:36 -0500)
committerDavid Goulet <dgoulet@efficios.com>
Mon, 18 Feb 2013 18:48:44 +0000 (13:48 -0500)
Before this change, applications were performing allocation and teardown
of their buffers. Applications therefore had ownership of the buffers.
The shm and wait fd were passed from applications, though sessiond, to
consumerd through unix sockets.

This change moves ownership of buffers from applications to consumer.
This will allow sharing buffers across many processes in a near future.
It will also facilitate implementation of periodical timers on the
consumer side, now that it has ownership of channels and buffers (also
called streams). This imply that file descriptors on shm and wakeup end
of the pipe are now passed from the consumerd to sessiond, then to
applications, through unix sockets. Then, applications "map" channel and
streams into their own memory space. Channel control structure is
actually a copy for each application, while streams are a shared memory
map (shm) between consumerd and all applications that write into it, and
have a wake up file descriptor on the application side.

Dependency on libuuid has been added to lttng-tools, since the UUID is
needed at channel and buffer allocation.

Note that this is only for UST so the kernel buffers are still created
in the session daemon then passed to the consumer. There is basically no
change for the kernel other than adapting to the new communication data
structure.

This commit needs to be used along with commit named "Move LTTng-UST
buffer ownership from application to consumer" in lttng-ust.

Reviewed-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Signed-off-by: David Goulet <dgoulet@efficios.com>
32 files changed:
configure.ac
src/bin/lttng-sessiond/cmd.c
src/bin/lttng-sessiond/cmd.h
src/bin/lttng-sessiond/consumer.c
src/bin/lttng-sessiond/consumer.h
src/bin/lttng-sessiond/kernel-consumer.c
src/bin/lttng-sessiond/kernel-consumer.h
src/bin/lttng-sessiond/lttng-ust-abi.h
src/bin/lttng-sessiond/lttng-ust-ctl.h
src/bin/lttng-sessiond/main.c
src/bin/lttng-sessiond/trace-kernel.c
src/bin/lttng-sessiond/trace-kernel.h
src/bin/lttng-sessiond/trace-ust.h
src/bin/lttng-sessiond/ust-app.c
src/bin/lttng-sessiond/ust-app.h
src/bin/lttng-sessiond/ust-consumer.c
src/bin/lttng-sessiond/ust-consumer.h
src/common/Makefile.am
src/common/compat/Makefile.am
src/common/compat/tid.h [new file with mode: 0644]
src/common/compat/uuid.h [new file with mode: 0644]
src/common/consumer.c
src/common/consumer.h
src/common/defaults.h
src/common/error.h
src/common/kernel-consumer/kernel-consumer.c
src/common/kernel-consumer/kernel-consumer.h
src/common/relayd/relayd.c
src/common/sessiond-comm/sessiond-comm.h
src/common/ust-consumer/ust-consumer.c
src/common/ust-consumer/ust-consumer.h
tests/tools/streaming/runall

index 91e76f06a98700d2a9deb36c29a650176e62c4ab..71fbb6034e72306f1b40b4b6394ec93a66c1efdd 100644 (file)
@@ -108,6 +108,27 @@ AC_CHECK_LIB([popt], [poptGetContext], [],
        [AC_MSG_ERROR([Cannot find libpopt. Use [LDFLAGS]=-Ldir to specify its location.])]
 )
 
+# Check for libuuid
+AC_CHECK_LIB([uuid], [uuid_generate],
+[
+       AC_DEFINE_UNQUOTED([LTTNG_HAVE_LIBUUID], 1, [Has libuuid support.])
+       have_libuuid=yes
+],
+[
+       # libuuid not found, check for uuid_create in libc.
+       AC_CHECK_LIB([c], [uuid_create],
+       [
+               AC_DEFINE_UNQUOTED([LTTNG_HAVE_LIBC_UUID], 1, [Has libc uuid support.])
+               have_libc_uuid=yes
+       ],
+       [
+               AC_MSG_ERROR([Cannot find libuuid uuid_generate nor libc uuid_create. Use [LDFLAGS]=-Ldir to specify their location.])
+       ])
+]
+)
+AM_CONDITIONAL([LTTNG_BUILD_WITH_LIBUUID], [test "x$have_libuuid" = "xyes"])
+AM_CONDITIONAL([LTTNG_BUILD_WITH_LIBC_UUID], [test "x$have_libc_uuid" = "xyes"])
+
 # URCU library version needed or newer
 liburcu_version=">= 0.7.2"
 
index bdc9c9acab52a1cbe893efe1faaa50198b27206e..150a2dfe4675bfbfaf90d5e30b51d54ab6eb284d 100644 (file)
@@ -376,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) {
@@ -484,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;
@@ -538,6 +546,17 @@ static int send_consumer_relayd_socket(int domain, struct ltt_session *session,
        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) {
                /*
@@ -550,17 +569,6 @@ 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_sock, sock,
                        consumer, relayd_uri->stype, session->id);
@@ -571,9 +579,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;
@@ -589,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;
 }
 
@@ -607,7 +623,7 @@ 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, sock);
                if (ret != LTTNG_OK) {
@@ -616,7 +632,7 @@ static int send_consumer_relayd_sockets(int domain,
        }
 
        /* 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, sock);
                if (ret != LTTNG_OK) {
@@ -633,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;
@@ -1481,12 +1497,6 @@ 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) {
                ret = start_kernel_session(ksession, kernel_tracer_fd);
@@ -1660,44 +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
-                * 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)) {
-                       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);
-                       pthread_mutex_unlock(socket->lock);
-                       if (ret != LTTNG_OK) {
-                               rcu_read_unlock();
-                               goto error;
-                       }
-               }
-               rcu_read_unlock();
        }
 
        /* All good! */
index c8619a0f6e270c8c58ad665d5ff148da153a13fa..59216138ce4a53fa7fe1ea84800fdacb5b05416b 100644 (file)
@@ -66,6 +66,7 @@ int cmd_disable_consumer(int domain, struct ltt_session *session);
 int cmd_enable_consumer(int domain, struct ltt_session *session);
 int cmd_set_consumer_uri(int domain, struct ltt_session *session,
                size_t nb_uri, struct lttng_uri *uris);
+int cmd_setup_relayd(struct ltt_session *session);
 
 /* Listing commands */
 ssize_t cmd_list_domains(struct ltt_session *session,
index 344c6627dbbe507c01ba483c7ba46c8f2577b57c..ff5360aae755aa082ec6517cb6c4f10df99c9d9d 100644 (file)
@@ -60,13 +60,55 @@ int consumer_recv_status_reply(struct consumer_socket *sock)
                ret = 0;
        } else {
                ret = -reply.ret_code;
-               DBG("Consumer ret code %d", reply.ret_code);
+               DBG("Consumer ret code %d", ret);
        }
 
 end:
        return ret;
 }
 
+/*
+ * Once the ASK_CHANNEL command is sent to the consumer, the channel
+ * information are sent back. This call receives that data and populates key
+ * and stream_count.
+ *
+ * On success return 0 and both key and stream_count are set. On error, a
+ * negative value is sent back and both parameters are untouched.
+ */
+int consumer_recv_status_channel(struct consumer_socket *sock,
+               unsigned long *key, unsigned int *stream_count)
+{
+       int ret;
+       struct lttcomm_consumer_status_channel reply;
+
+       assert(sock);
+       assert(stream_count);
+       assert(key);
+
+       ret = lttcomm_recv_unix_sock(sock->fd, &reply, sizeof(reply));
+       if (ret <= 0) {
+               if (ret == 0) {
+                       /* Orderly shutdown. Don't return 0 which means success. */
+                       ret = -1;
+               }
+               /* The above call will print a PERROR on error. */
+               DBG("Fail to receive status reply on sock %d", sock->fd);
+               goto end;
+       }
+
+       /* An error is possible so don't touch the key and stream_count. */
+       if (reply.ret_code != LTTNG_OK) {
+               ret = -1;
+               goto end;
+       }
+
+       *key = reply.key;
+       *stream_count = reply.stream_count;
+
+end:
+       return ret;
+}
+
 /*
  * Send destroy relayd command to consumer.
  *
@@ -81,7 +123,7 @@ int consumer_send_destroy_relayd(struct consumer_socket *sock,
        assert(consumer);
        assert(sock);
 
-       DBG2("Sending destroy relayd command to consumer...");
+       DBG2("Sending destroy relayd command to consumer sock %d", sock->fd);
 
        /* Bail out if consumer is disabled */
        if (!consumer->enabled) {
@@ -511,6 +553,32 @@ error:
        return ret;
 }
 
+/*
+ * Consumer send communication message structure to consumer.
+ */
+int consumer_send_msg(struct consumer_socket *sock,
+               struct lttcomm_consumer_msg *msg)
+{
+       int ret;
+
+       assert(msg);
+       assert(sock);
+       assert(sock->fd >= 0);
+
+       ret = lttcomm_send_unix_sock(sock->fd, msg,
+                       sizeof(struct lttcomm_consumer_msg));
+       if (ret < 0) {
+               /* The above call will print a PERROR on error. */
+               DBG("Error when sending consumer channel on sock %d", sock->fd);
+               goto error;
+       }
+
+       ret = consumer_recv_status_reply(sock);
+
+error:
+       return ret;
+}
+
 /*
  * Consumer send channel communication message structure to consumer.
  */
@@ -537,30 +605,94 @@ error:
        return ret;
 }
 
+/*
+ * Populate the given consumer msg structure with the ask_channel command
+ * information.
+ */
+void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
+               uint64_t subbuf_size,
+               uint64_t num_subbuf,
+               int overwrite,
+               unsigned int switch_timer_interval,
+               unsigned int read_timer_interval,
+               int output,
+               int type,
+               uint64_t session_id,
+               const char *pathname,
+               const char *name,
+               uid_t uid,
+               gid_t gid,
+               int relayd_id,
+               unsigned long key,
+               unsigned char *uuid)
+{
+       assert(msg);
+
+       /* Zeroed structure */
+       memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
+
+       msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
+       msg->u.ask_channel.subbuf_size = subbuf_size;
+       msg->u.ask_channel.num_subbuf = num_subbuf ;
+       msg->u.ask_channel.overwrite = overwrite;
+       msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
+       msg->u.ask_channel.read_timer_interval = read_timer_interval;
+       msg->u.ask_channel.output = output;
+       msg->u.ask_channel.type = type;
+       msg->u.ask_channel.session_id = session_id;
+       msg->u.ask_channel.uid = uid;
+       msg->u.ask_channel.gid = gid;
+       msg->u.ask_channel.relayd_id = relayd_id;
+       msg->u.ask_channel.key = key;
+
+       memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
+
+       strncpy(msg->u.ask_channel.pathname, pathname,
+                       sizeof(msg->u.ask_channel.pathname));
+       msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
+
+       strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
+       msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
+}
+
 /*
  * Init channel communication message structure.
  */
 void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
                enum lttng_consumer_command cmd,
                int channel_key,
-               uint64_t max_sb_size,
-               uint64_t mmap_len,
+               uint64_t session_id,
+               const char *pathname,
+               uid_t uid,
+               gid_t gid,
+               int relayd_id,
                const char *name,
-               unsigned int nb_init_streams)
+               unsigned int nb_init_streams,
+               enum lttng_event_output output,
+               int type)
 {
        assert(msg);
 
-       /* TODO: Args validation */
-
        /* Zeroed structure */
        memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 
        /* Send channel */
        msg->cmd_type = cmd;
        msg->u.channel.channel_key = channel_key;
-       msg->u.channel.max_sb_size = max_sb_size;
-       msg->u.channel.mmap_len = mmap_len;
+       msg->u.channel.session_id = session_id;
+       msg->u.channel.uid = uid;
+       msg->u.channel.gid = gid;
+       msg->u.channel.relayd_id = relayd_id;
        msg->u.channel.nb_init_streams = nb_init_streams;
+       msg->u.channel.output = output;
+       msg->u.channel.type = type;
+
+       strncpy(msg->u.channel.pathname, pathname,
+                       sizeof(msg->u.channel.pathname));
+       msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
+
+       strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
+       msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
 }
 
 /*
@@ -570,39 +702,16 @@ void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
                enum lttng_consumer_command cmd,
                int channel_key,
                int stream_key,
-               uint32_t state,
-               enum lttng_event_output output,
-               uint64_t mmap_len,
-               uid_t uid,
-               gid_t gid,
-               int net_index,
-               unsigned int metadata_flag,
-               const char *name,
-               const char *pathname,
-               unsigned int session_id)
+               int cpu)
 {
        assert(msg);
 
        memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 
-       /* TODO: Args validation */
-
        msg->cmd_type = cmd;
        msg->u.stream.channel_key = channel_key;
        msg->u.stream.stream_key = stream_key;
-       msg->u.stream.state = state;
-       msg->u.stream.output = output;
-       msg->u.stream.mmap_len = mmap_len;
-       msg->u.stream.uid = uid;
-       msg->u.stream.gid = gid;
-       msg->u.stream.net_index = net_index;
-       msg->u.stream.metadata_flag = metadata_flag;
-       msg->u.stream.session_id = (uint64_t) session_id;
-       strncpy(msg->u.stream.name, name, sizeof(msg->u.stream.name));
-       msg->u.stream.name[sizeof(msg->u.stream.name) - 1] = '\0';
-       strncpy(msg->u.stream.path_name, pathname,
-                       sizeof(msg->u.stream.path_name));
-       msg->u.stream.path_name[sizeof(msg->u.stream.path_name) - 1] = '\0';
+       msg->u.stream.cpu = cpu;
 }
 
 /*
@@ -617,29 +726,7 @@ int consumer_send_stream(struct consumer_socket *sock,
        assert(msg);
        assert(dst);
        assert(sock);
-
-       switch (dst->type) {
-       case CONSUMER_DST_NET:
-               /* Consumer should send the stream on the network. */
-               msg->u.stream.net_index = dst->net_seq_index;
-               break;
-       case CONSUMER_DST_LOCAL:
-               /* Add stream file name to stream path */
-               strncat(msg->u.stream.path_name, "/",
-                               sizeof(msg->u.stream.path_name) -
-                               strlen(msg->u.stream.path_name) - 1);
-               strncat(msg->u.stream.path_name, msg->u.stream.name,
-                               sizeof(msg->u.stream.path_name) -
-                               strlen(msg->u.stream.path_name) - 1);
-               msg->u.stream.path_name[sizeof(msg->u.stream.path_name) - 1] = '\0';
-               /* Indicate that the stream is NOT network */
-               msg->u.stream.net_index = -1;
-               break;
-       default:
-               ERR("Consumer unknown output type (%d)", dst->type);
-               ret = -1;
-               goto error;
-       }
+       assert(fds);
 
        /* Send on socket */
        ret = lttcomm_send_unix_sock(sock->fd, msg,
index 11c98773ee363de996ff03d6d158695c4f6e67c7..af337baa386b61f81e587609a03d3e7ae800d1a2 100644 (file)
@@ -45,6 +45,10 @@ struct consumer_socket {
         */
        unsigned int registered;
 
+       /* Flag if network sockets were sent to the consumer. */
+       unsigned int control_sock_sent;
+       unsigned int data_sock_sent;
+
        struct lttng_ht_node_ulong node;
 };
 
@@ -107,10 +111,6 @@ struct consumer_net {
 
        /* Data path for network streaming. */
        struct lttng_uri data;
-
-       /* Flag if network sockets were sent to the consumer. */
-       unsigned int control_sock_sent;
-       unsigned int data_sock_sent;
 };
 
 /*
@@ -161,6 +161,8 @@ void consumer_destroy_output(struct consumer_output *obj);
 int consumer_set_network_uri(struct consumer_output *obj,
                struct lttng_uri *uri);
 int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd);
+int consumer_send_msg(struct consumer_socket *sock,
+               struct lttcomm_consumer_msg *msg);
 int consumer_send_stream(struct consumer_socket *sock,
                struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
                int *fds, size_t nb_fd);
@@ -172,33 +174,47 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
 int consumer_send_destroy_relayd(struct consumer_socket *sock,
                struct consumer_output *consumer);
 int consumer_recv_status_reply(struct consumer_socket *sock);
+int consumer_recv_status_channel(struct consumer_socket *sock,
+               unsigned long *key, unsigned int *stream_count);
 void consumer_output_send_destroy_relayd(struct consumer_output *consumer);
 int consumer_create_socket(struct consumer_data *data,
                struct consumer_output *output);
 int consumer_set_subdir(struct consumer_output *consumer,
                const char *session_name);
 
+void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
+               uint64_t subbuf_size,
+               uint64_t num_subbuf,
+               int overwrite,
+               unsigned int switch_timer_interval,
+               unsigned int read_timer_interval,
+               int output,
+               int type,
+               uint64_t session_id,
+               const char *pathname,
+               const char *name,
+               uid_t uid,
+               gid_t gid,
+               int relayd_id,
+               unsigned long key,
+               unsigned char *uuid);
 void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
                enum lttng_consumer_command cmd,
                int channel_key,
                int stream_key,
-               uint32_t state,
-               enum lttng_event_output output,
-               uint64_t mmap_len,
-               uid_t uid,
-               gid_t gid,
-               int net_index,
-               unsigned int metadata_flag,
-               const char *name,
-               const char *pathname,
-               unsigned int session_id);
+               int cpu);
 void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
                enum lttng_consumer_command cmd,
                int channel_key,
-               uint64_t max_sb_size,
-               uint64_t mmap_len,
+               uint64_t session_id,
+               const char *pathname,
+               uid_t uid,
+               gid_t gid,
+               int relayd_id,
                const char *name,
-               unsigned int nb_init_streams);
+               unsigned int nb_init_streams,
+               enum lttng_event_output output,
+               int type);
 int consumer_is_data_pending(unsigned int id,
                struct consumer_output *consumer);
 
index bdf9b4098be5edef16f3f1d9ad076637079304ec..837afdc8333faf71de79dfe56bbff31ae44a236a 100644 (file)
  * Sending a single channel to the consumer with command ADD_CHANNEL.
  */
 int kernel_consumer_add_channel(struct consumer_socket *sock,
-               struct ltt_kernel_channel *channel)
+               struct ltt_kernel_channel *channel, struct ltt_kernel_session *session)
 {
        int ret;
+       char tmp_path[PATH_MAX];
+       const char *pathname;
        struct lttcomm_consumer_msg lkm;
+       struct consumer_output *consumer;
 
        /* Safety net */
        assert(channel);
+       assert(session);
+       assert(session->consumer);
+
+       consumer = session->consumer;
 
        DBG("Kernel consumer adding channel %s to kernel consumer",
                        channel->channel->name);
 
+       /* Get the right path name destination */
+       if (consumer->type == CONSUMER_DST_LOCAL) {
+               /* Set application path to the destination path */
+               ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
+                               consumer->dst.trace_path, consumer->subdir);
+               if (ret < 0) {
+                       PERROR("snprintf metadata path");
+                       goto error;
+               }
+               pathname = tmp_path;
+
+               /* Create directory */
+               ret = run_as_mkdir_recursive(pathname, S_IRWXU | S_IRWXG,
+                               session->uid, session->gid);
+               if (ret < 0) {
+                       if (ret != -EEXIST) {
+                               ERR("Trace directory creation error");
+                               goto error;
+                       }
+               }
+               DBG3("Kernel local consumer tracefile path: %s", pathname);
+       } else {
+               ret = snprintf(tmp_path, sizeof(tmp_path), "%s", consumer->subdir);
+               if (ret < 0) {
+                       PERROR("snprintf metadata path");
+                       goto error;
+               }
+               pathname = tmp_path;
+               DBG3("Kernel network consumer subdir path: %s", pathname);
+       }
+
        /* Prep channel message structure */
        consumer_init_channel_comm_msg(&lkm,
                        LTTNG_CONSUMER_ADD_CHANNEL,
                        channel->fd,
-                       channel->channel->attr.subbuf_size,
-                       0, /* Kernel */
+                       session->id,
+                       pathname,
+                       session->uid,
+                       session->gid,
+                       consumer->net_seq_index,
                        channel->channel->name,
-                       channel->stream_count);
+                       channel->stream_count,
+                       channel->channel->attr.output,
+                       CONSUMER_CHANNEL_TYPE_DATA);
 
        health_code_update();
 
@@ -122,10 +165,15 @@ int kernel_consumer_add_metadata(struct consumer_socket *sock,
        consumer_init_channel_comm_msg(&lkm,
                        LTTNG_CONSUMER_ADD_CHANNEL,
                        session->metadata->fd,
-                       session->metadata->conf->attr.subbuf_size,
-                       0, /* for kernel */
+                       session->id,
+                       pathname,
+                       session->uid,
+                       session->gid,
+                       consumer->net_seq_index,
                        DEFAULT_METADATA_NAME,
-                       1);
+                       1,
+                       DEFAULT_KERNEL_CHANNEL_OUTPUT,
+                       CONSUMER_CHANNEL_TYPE_METADATA);
 
        health_code_update();
 
@@ -141,16 +189,7 @@ int kernel_consumer_add_metadata(struct consumer_socket *sock,
                        LTTNG_CONSUMER_ADD_STREAM,
                        session->metadata->fd,
                        session->metadata_stream_fd,
-                       LTTNG_CONSUMER_ACTIVE_STREAM,
-                       DEFAULT_KERNEL_CHANNEL_OUTPUT,
-                       0, /* Kernel */
-                       session->uid,
-                       session->gid,
-                       consumer->net_seq_index,
-                       1, /* Metadata flag set */
-                       DEFAULT_METADATA_NAME,
-                       pathname,
-                       session->id);
+                       0);     /* CPU: 0 for metadata. */
 
        health_code_update();
 
@@ -175,8 +214,6 @@ int kernel_consumer_add_stream(struct consumer_socket *sock,
                struct ltt_kernel_session *session)
 {
        int ret;
-       char tmp_path[PATH_MAX];
-       const char *pathname;
        struct lttcomm_consumer_msg lkm;
        struct consumer_output *consumer;
 
@@ -192,41 +229,12 @@ int kernel_consumer_add_stream(struct consumer_socket *sock,
        /* Get consumer output pointer */
        consumer = session->consumer;
 
-       /* Get the right path name destination */
-       if (consumer->type == CONSUMER_DST_LOCAL) {
-               /* Set application path to the destination path */
-               ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
-                               consumer->dst.trace_path, consumer->subdir);
-               if (ret < 0) {
-                       PERROR("snprintf stream path");
-                       goto error;
-               }
-               pathname = tmp_path;
-               DBG3("Kernel local consumer tracefile path: %s", pathname);
-       } else {
-               ret = snprintf(tmp_path, sizeof(tmp_path), "%s", consumer->subdir);
-               if (ret < 0) {
-                       PERROR("snprintf stream path");
-                       goto error;
-               }
-               pathname = tmp_path;
-               DBG3("Kernel network consumer subdir path: %s", pathname);
-       }
-
        /* Prep stream consumer message */
-       consumer_init_stream_comm_msg(&lkm, LTTNG_CONSUMER_ADD_STREAM,
+       consumer_init_stream_comm_msg(&lkm,
+                       LTTNG_CONSUMER_ADD_STREAM,
                        channel->fd,
                        stream->fd,
-                       stream->state,
-                       channel->channel->attr.output,
-                       0, /* Kernel */
-                       session->uid,
-                       session->gid,
-                       consumer->net_seq_index,
-                       0, /* Metadata flag unset */
-                       stream->name,
-                       pathname,
-                       session->id);
+                       stream->cpu);
 
        health_code_update();
 
@@ -266,7 +274,7 @@ int kernel_consumer_send_channel_stream(struct consumer_socket *sock,
        DBG("Sending streams of channel %s to kernel consumer",
                        channel->channel->name);
 
-       ret = kernel_consumer_add_channel(sock, channel);
+       ret = kernel_consumer_add_channel(sock, channel, session);
        if (ret < 0) {
                goto error;
        }
index 07d7436a14d8efe41c62b0479f1d440ffd4acff4..b9a424e541924e608cd9ba0a224f50eeb2772ca7 100644 (file)
@@ -35,4 +35,4 @@ int kernel_consumer_add_metadata(struct consumer_socket *sock,
                struct ltt_kernel_session *session);
 
 int kernel_consumer_add_channel(struct consumer_socket *sock,
-               struct ltt_kernel_channel *channel);
+               struct ltt_kernel_channel *channel, struct ltt_kernel_session *session);
index ef663ced23ca633991eede0c73a7ece811a16400..4ef04653cc517bebc3ec87890f03bc9dcc975622 100644 (file)
@@ -28,7 +28,7 @@
  */
 
 #include <stdint.h>
-#include <common/macros.h>
+#include <lttng/ust-compiler.h>
 
 #define LTTNG_UST_SYM_NAME_LEN 256
 
@@ -57,6 +57,11 @@ enum lttng_ust_output {
        LTTNG_UST_MMAP          = 0,
 };
 
+enum lttng_ust_chan_type {
+       LTTNG_UST_CHAN_PER_CPU = 0,
+       LTTNG_UST_CHAN_METADATA = 1,
+};
+
 struct lttng_ust_tracer_version {
        uint32_t major;
        uint32_t minor;
@@ -64,24 +69,29 @@ struct lttng_ust_tracer_version {
 } LTTNG_PACKED;
 
 #define LTTNG_UST_CHANNEL_PADDING      LTTNG_UST_SYM_NAME_LEN + 32
+/*
+ * Given that the consumerd is limited to 64k file descriptors, we
+ * cannot expect much more than 1MB channel structure size. This size is
+ * depends on the number of streams within a channel, which depends on
+ * the number of possible CPUs on the system.
+ */
+#define LTTNG_UST_CHANNEL_DATA_MAX_LEN 1048576U
 struct lttng_ust_channel {
-       uint64_t subbuf_size;                   /* in bytes */
-       uint64_t num_subbuf;
-       int overwrite;                          /* 1: overwrite, 0: discard */
-       unsigned int switch_timer_interval;     /* usecs */
-       unsigned int read_timer_interval;       /* usecs */
-       enum lttng_ust_output output;           /* output mode */
+       uint64_t len;
+       enum lttng_ust_chan_type type;
        char padding[LTTNG_UST_CHANNEL_PADDING];
+       char data[];    /* variable sized data */
 } LTTNG_PACKED;
 
-#define LTTNG_UST_STREAM_PADDING1      16
-#define LTTNG_UST_STREAM_PADDING2      LTTNG_UST_SYM_NAME_LEN + 32
+#define LTTNG_UST_STREAM_PADDING1      LTTNG_UST_SYM_NAME_LEN + 32
 struct lttng_ust_stream {
+       uint64_t len;           /* shm len */
+       uint32_t stream_nr;     /* stream number */
        char padding[LTTNG_UST_STREAM_PADDING1];
-
-       union {
-               char padding[LTTNG_UST_STREAM_PADDING2];
-       } u;
+       /*
+        * shm_fd and wakeup_fd are send over unix socket as file
+        * descriptors after this structure.
+        */
 } LTTNG_PACKED;
 
 #define LTTNG_UST_EVENT_PADDING1       16
@@ -157,13 +167,32 @@ struct lttng_ust_tracepoint_iter {
        char padding[LTTNG_UST_TRACEPOINT_ITER_PADDING];
 } LTTNG_PACKED;
 
-#define LTTNG_UST_OBJECT_DATA_PADDING          LTTNG_UST_SYM_NAME_LEN + 32
+enum lttng_ust_object_type {
+       LTTNG_UST_OBJECT_TYPE_UNKNOWN = -1,
+       LTTNG_UST_OBJECT_TYPE_CHANNEL = 0,
+       LTTNG_UST_OBJECT_TYPE_STREAM = 1,
+};
+
+#define LTTNG_UST_OBJECT_DATA_PADDING1         32
+#define LTTNG_UST_OBJECT_DATA_PADDING2         LTTNG_UST_SYM_NAME_LEN + 32
+
 struct lttng_ust_object_data {
-       uint64_t memory_map_size;
+       enum lttng_ust_object_type type;
        int handle;
-       int shm_fd;
-       int wait_fd;
-       char padding[LTTNG_UST_OBJECT_DATA_PADDING];
+       uint64_t size;
+       char padding1[LTTNG_UST_OBJECT_DATA_PADDING1];
+       union {
+               struct {
+                       void *data;
+                       enum lttng_ust_chan_type type;
+               } channel;
+               struct {
+                       int shm_fd;
+                       int wakeup_fd;
+                       uint32_t stream_nr;
+               } stream;
+               char padding2[LTTNG_UST_OBJECT_DATA_PADDING2];
+       } u;
 } LTTNG_PACKED;
 
 enum lttng_ust_calibrate_type {
@@ -245,14 +274,11 @@ struct lttng_ust_obj;
 
 union ust_args {
        struct {
-               int *shm_fd;
-               int *wait_fd;
-               uint64_t *memory_map_size;
+               void *chan_data;
        } channel;
        struct {
-               int *shm_fd;
-               int *wait_fd;
-               uint64_t *memory_map_size;
+               int shm_fd;
+               int wakeup_fd;
        } stream;
        struct {
                struct lttng_ust_field_iter entry;
index 9baf443f3fd443353c99c054c0f00bd756c86395..cd833b826c589f4f59a88ec554a823dfe0cc3e29 100644 (file)
@@ -1,10 +1,10 @@
 /*
  * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
- *                      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ * Copyright (C) 2011-2013 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
  *
  * This program is free software; you can redistribute it and/or modify
- * it under the terms of the GNU General Public License, version 2 only,
- * as published by the Free Software Foundation.
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; version 2 of the License only.
  *
  * This program is distributed in the hope that it will be useful,
  * but WITHOUT ANY WARRANTY; without even the implied warranty of
 #ifndef _LTTNG_UST_CTL_H
 #define _LTTNG_UST_CTL_H
 
-#include "lttng-ust-abi.h"
+#include <lttng/ust-abi.h>
 
+#ifndef LTTNG_PACKED
+#define LTTNG_PACKED __attribute__((packed))
+#endif
+
+#ifndef LTTNG_UST_UUID_LEN
+#define LTTNG_UST_UUID_LEN     16
+#endif
+
+struct lttng_ust_shm_handle;
+struct lttng_ust_lib_ring_buffer;
+
+struct ustctl_consumer_channel_attr {
+       enum lttng_ust_chan_type type;
+       uint64_t subbuf_size;                   /* bytes */
+       uint64_t num_subbuf;                    /* power of 2 */
+       int overwrite;                          /* 1: overwrite, 0: discard */
+       unsigned int switch_timer_interval;     /* usec */
+       unsigned int read_timer_interval;       /* usec */
+       enum lttng_ust_output output;           /* splice, mmap */
+       unsigned char uuid[LTTNG_UST_UUID_LEN]; /* Trace session unique ID */
+} LTTNG_PACKED;
+
+/*
+ * API used by sessiond.
+ */
+
+/*
+ * Error values: all the following functions return:
+ * >= 0: Success (LTTNG_UST_OK)
+ * < 0: error code.
+ */
 int ustctl_register_done(int sock);
 int ustctl_create_session(int sock);
-int ustctl_open_metadata(int sock, int session_handle,
-               struct lttng_ust_channel_attr *chops,
-               struct lttng_ust_object_data **metadata_data);
-int ustctl_create_channel(int sock, int session_handle,
-               struct lttng_ust_channel_attr *chops,
-               struct lttng_ust_object_data **channel_data);
-int ustctl_create_stream(int sock, struct lttng_ust_object_data *channel_data,
-               struct lttng_ust_object_data **stream_data);
 int ustctl_create_event(int sock, struct lttng_ust_event *ev,
                struct lttng_ust_object_data *channel_data,
                struct lttng_ust_object_data **event_data);
@@ -50,9 +73,11 @@ int ustctl_stop_session(int sock, int handle);
  * error value.
  */
 int ustctl_tracepoint_list(int sock);
+
 /*
  * ustctl_tracepoint_list_get is used to iterate on the tp list
- * handle. End is iteration is reached when -ENOENT is returned.
+ * handle. End is iteration is reached when -LTTNG_UST_ERR_NOENT is
+ * returned.
  */
 int ustctl_tracepoint_list_get(int sock, int tp_list_handle,
                struct lttng_ust_tracepoint_iter *iter);
@@ -65,7 +90,8 @@ int ustctl_tracepoint_field_list(int sock);
 
 /*
  * ustctl_tracepoint_field_list_get is used to iterate on the tp field
- * list handle. End is iteration is reached when -ENOENT is returned.
+ * list handle. End is iteration is reached when -LTTNG_UST_ERR_NOENT is
+ * returned.
  */
 int ustctl_tracepoint_field_list_get(int sock, int tp_field_list_handle,
                struct lttng_ust_field_iter *iter);
@@ -75,78 +101,90 @@ int ustctl_wait_quiescent(int sock);
 
 int ustctl_sock_flush_buffer(int sock, struct lttng_ust_object_data *object);
 
-/* not implemented yet */
 int ustctl_calibrate(int sock, struct lttng_ust_calibrate *calibrate);
 
+/* Release object created by members of this API. */
+int ustctl_release_object(int sock, struct lttng_ust_object_data *data);
+/* Release handle returned by create session. */
+int ustctl_release_handle(int sock, int handle);
+
+int ustctl_recv_channel_from_consumer(int sock,
+               struct lttng_ust_object_data **channel_data);
+int ustctl_recv_stream_from_consumer(int sock,
+               struct lttng_ust_object_data **stream_data);
+int ustctl_send_channel_to_ust(int sock, int session_handle,
+               struct lttng_ust_object_data *channel_data);
+int ustctl_send_stream_to_ust(int sock,
+               struct lttng_ust_object_data *channel_data,
+               struct lttng_ust_object_data *stream_data);
+
 /*
- * Map channel lttng_ust_shm_handle and add streams. Typically performed by the
- * consumer to map the objects into its memory space.
+ * API used by consumer.
  */
-struct lttng_ust_shm_handle *ustctl_map_channel(struct lttng_ust_object_data *chan_data);
-int ustctl_add_stream(struct lttng_ust_shm_handle *lttng_ust_shm_handle,
-               struct lttng_ust_object_data *stream_data);
+
+struct ustctl_consumer_channel;
+struct ustctl_consumer_stream;
+struct ustctl_consumer_channel_attr;
+
+struct ustctl_consumer_channel *
+       ustctl_create_channel(struct ustctl_consumer_channel_attr *attr);
 /*
- * Note: the lttng_ust_object_data from which the lttng_ust_shm_handle is derived can only
- * be released after unmapping the handle.
+ * Each stream created needs to be destroyed before calling
+ * ustctl_destroy_channel().
  */
-void ustctl_unmap_channel(struct lttng_ust_shm_handle *lttng_ust_shm_handle);
+void ustctl_destroy_channel(struct ustctl_consumer_channel *chan);
 
-/* Buffer operations */
+int ustctl_send_channel_to_sessiond(int sock,
+               struct ustctl_consumer_channel *channel);
+/*
+ * Send a NULL stream to finish iteration over all streams of a given
+ * channel.
+ */
+int ustctl_send_stream_to_sessiond(int sock,
+               struct ustctl_consumer_stream *stream);
+int ustctl_stream_close_wakeup_fd(struct ustctl_consumer_stream *stream);
 
-struct lttng_ust_shm_handle;
-struct lttng_ust_lib_ring_buffer;
+/* Create/destroy stream buffers for read */
+struct ustctl_consumer_stream *
+       ustctl_create_stream(struct ustctl_consumer_channel *channel,
+                       int cpu);
+void ustctl_destroy_stream(struct ustctl_consumer_stream *stream);
 
-/* Open/close stream buffers for read */
-struct lttng_ust_lib_ring_buffer *ustctl_open_stream_read(struct lttng_ust_shm_handle *handle,
-               int cpu);
-void ustctl_close_stream_read(struct lttng_ust_shm_handle *handle,
-                struct lttng_ust_lib_ring_buffer *buf);
+int ustctl_get_wait_fd(struct ustctl_consumer_stream *stream);
+int ustctl_get_wakeup_fd(struct ustctl_consumer_stream *stream);
 
 /* For mmap mode, readable without "get" operation */
-int ustctl_get_mmap_len(struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf,
+int ustctl_get_mmap_len(struct ustctl_consumer_stream *stream,
                unsigned long *len);
-int ustctl_get_max_subbuf_size(struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf,
+int ustctl_get_max_subbuf_size(struct ustctl_consumer_stream *stream,
                unsigned long *len);
 
 /*
  * For mmap mode, operate on the current packet (between get/put or
  * get_next/put_next).
  */
-void *ustctl_get_mmap_base(struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf);
-int ustctl_get_mmap_read_offset(struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf, unsigned long *off);
-int ustctl_get_subbuf_size(struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf, unsigned long *len);
-int ustctl_get_padded_subbuf_size(struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf, unsigned long *len);
-int ustctl_get_next_subbuf(struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf);
-int ustctl_put_next_subbuf(struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf);
+void *ustctl_get_mmap_base(struct ustctl_consumer_stream *stream);
+int ustctl_get_mmap_read_offset(struct ustctl_consumer_stream *stream,
+               unsigned long *off);
+int ustctl_get_subbuf_size(struct ustctl_consumer_stream *stream,
+               unsigned long *len);
+int ustctl_get_padded_subbuf_size(struct ustctl_consumer_stream *stream,
+               unsigned long *len);
+int ustctl_get_next_subbuf(struct ustctl_consumer_stream *stream);
+int ustctl_put_next_subbuf(struct ustctl_consumer_stream *stream);
 
 /* snapshot */
 
-int ustctl_snapshot(struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf);
-int ustctl_snapshot_get_consumed(struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf, unsigned long *pos);
-int ustctl_snapshot_get_produced(struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf, unsigned long *pos);
-int ustctl_get_subbuf(struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf, unsigned long *pos);
-int ustctl_put_subbuf(struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf);
-
-void ustctl_flush_buffer(struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf,
-               int producer_active);
+int ustctl_snapshot(struct ustctl_consumer_stream *stream);
+int ustctl_snapshot_get_consumed(struct ustctl_consumer_stream *stream,
+               unsigned long *pos);
+int ustctl_snapshot_get_produced(struct ustctl_consumer_stream *stream,
+               unsigned long *pos);
+int ustctl_get_subbuf(struct ustctl_consumer_stream *stream,
+               unsigned long *pos);
+int ustctl_put_subbuf(struct ustctl_consumer_stream *stream);
 
-/* Release object created by members of this API */
-int ustctl_release_object(int sock, struct lttng_ust_object_data *data);
-/* Release handle returned by create session. */
-int ustctl_release_handle(int sock, int handle);
+void ustctl_flush_buffer(struct ustctl_consumer_stream *stream,
+               int producer_active);
 
 #endif /* _LTTNG_UST_CTL_H */
index 84b3f20ed5420b03210e8c80c4f638f4f5072d3b..fc0eb5cb6bbed1ef3b4727eb74a6135b0347fb4c 100644 (file)
@@ -664,14 +664,13 @@ error:
 }
 
 /*
- * For each tracing session, update newly registered apps.
+ * For each tracing session, update newly registered apps. The session list
+ * lock MUST be acquired before calling this.
  */
 static void update_ust_app(int app_sock)
 {
        struct ltt_session *sess, *stmp;
 
-       session_lock_list();
-
        /* For all tracing session(s) */
        cds_list_for_each_entry_safe(sess, stmp, &session_list_ptr->head, list) {
                session_lock(sess);
@@ -680,8 +679,6 @@ static void update_ust_app(int app_sock)
                }
                session_unlock(sess);
        }
-
-       session_unlock_list();
 }
 
 /*
@@ -1179,12 +1176,22 @@ static void *thread_manage_apps(void *data)
 
                                        health_code_update();
 
+                                       /*
+                                        * @session_lock
+                                        * Lock the global session list so from the register up to
+                                        * the registration done message, no thread can see the
+                                        * application and change its state.
+                                        */
+                                       session_lock_list();
+
                                        /* Register applicaton to the session daemon */
                                        ret = ust_app_register(&ust_cmd.reg_msg,
                                                        ust_cmd.sock);
                                        if (ret == -ENOMEM) {
+                                               session_unlock_list();
                                                goto error;
                                        } else if (ret < 0) {
+                                               session_unlock_list();
                                                break;
                                        }
 
@@ -1220,6 +1227,7 @@ static void *thread_manage_apps(void *data)
                                                ret = lttng_poll_add(&events, ust_cmd.sock,
                                                                LPOLLERR & LPOLLHUP & LPOLLRDHUP);
                                                if (ret < 0) {
+                                                       session_unlock_list();
                                                        goto error;
                                                }
 
@@ -1232,6 +1240,7 @@ static void *thread_manage_apps(void *data)
                                                DBG("Apps with sock %d added to poll set",
                                                                ust_cmd.sock);
                                        }
+                                       session_unlock_list();
 
                                        health_code_update();
 
@@ -2476,6 +2485,21 @@ skip_domain:
                }
        }
 
+       /*
+        * Send relayd information to consumer as soon as we have a domain and a
+        * session defined.
+        */
+       if (cmd_ctx->session && need_domain) {
+               /*
+                * Setup relayd if not done yet. If the relayd information was already
+                * sent to the consumer, this call will gracefully return.
+                */
+               ret = cmd_setup_relayd(cmd_ctx->session);
+               if (ret != LTTNG_OK) {
+                       goto error;
+               }
+       }
+
        /* Process by command type */
        switch (cmd_ctx->lsm->cmd_type) {
        case LTTNG_ADD_CONTEXT:
index 53053f2067c6a8b20cc92f0b6a3019fb360c473b..48be065765b2a548892bd534cea6cb0ec8c1df10 100644 (file)
@@ -319,6 +319,7 @@ struct ltt_kernel_stream *trace_kernel_create_stream(const char *name,
        /* Init stream */
        lks->fd = -1;
        lks->state = 0;
+       lks->cpu = count;
 
        return lks;
 
index 9050c4d044dad7185572bdec2f09430e41b15f2f..66ca8dbdef5a9b8f5818855b6cb51c88820ef319 100644 (file)
@@ -79,6 +79,7 @@ struct ltt_kernel_metadata {
 struct ltt_kernel_stream {
        int fd;
        int state;
+       int cpu;
        /* Format is %s_%d respectively channel name and CPU number. */
        char name[DEFAULT_STREAM_NAME_LEN];
        struct cds_list_head list;
index 97f3135a2063adcf621d300faf139c60cd54fcdc..f5fa092629d60e7030ac0b92aba0d943228268ab 100644 (file)
@@ -54,7 +54,7 @@ struct ltt_ust_channel {
        unsigned int enabled;
        char name[LTTNG_UST_SYM_NAME_LEN];
        char pathname[PATH_MAX];
-       struct lttng_ust_channel attr;
+       struct lttng_ust_channel_attr attr;
        struct lttng_ht *ctx;
        struct lttng_ht *events;
        struct lttng_ht_node_str node;
@@ -65,7 +65,7 @@ struct ltt_ust_metadata {
        int handle;
        struct lttng_ust_object_data *obj;
        char pathname[PATH_MAX];              /* Trace file path name */
-       struct lttng_ust_channel attr;
+       struct lttng_ust_channel_attr attr;
        struct lttng_ust_object_data *stream_obj;
 };
 
@@ -172,8 +172,8 @@ struct ltt_ust_channel *trace_ust_find_channel_by_name(struct lttng_ht *ht,
 }
 
 static inline
-struct ltt_ust_session *trace_ust_create_session(char *path, pid_t pid,
-               struct lttng_domain *domain)
+struct ltt_ust_session *trace_ust_create_session(char *path,
+               unsigned int session_id)
 {
        return NULL;
 }
index 679cace8a79063a07368163b46d7def48704f7f0..aa188931d7688769bfd81fbf0677a75b031a6092 100644 (file)
 #include "ust-consumer.h"
 #include "ust-ctl.h"
 
+/* Next available channel key. */
+static unsigned long next_channel_key;
+
+/*
+ * Return the atomically incremented value of next_channel_key.
+ */
+static inline unsigned long get_next_channel_key(void)
+{
+       return uatomic_add_return(&next_channel_key, 1);
+}
+
+/*
+ * Return the consumer socket from the given consumer output with the right
+ * bitness. On error, returns NULL.
+ *
+ * The caller MUST acquire a rcu read side lock and keep it until the socket
+ * object reference is not needed anymore.
+ */
+static struct consumer_socket *find_consumer_socket_by_bitness(int bits,
+               struct consumer_output *consumer)
+{
+       int consumer_fd;
+       struct consumer_socket *socket = NULL;
+
+       switch (bits) {
+       case 64:
+               consumer_fd = uatomic_read(&ust_consumerd64_fd);
+               break;
+       case 32:
+               consumer_fd = uatomic_read(&ust_consumerd32_fd);
+               break;
+       default:
+               assert(0);
+               goto end;
+       }
+
+       socket = consumer_find_socket(consumer_fd, consumer);
+
+end:
+       return socket;
+}
+
 /*
  * Match function for the hash table lookup.
  *
@@ -127,8 +169,16 @@ static void add_unique_ust_app_event(struct lttng_ht *ht,
 static
 void delete_ust_app_ctx(int sock, struct ust_app_ctx *ua_ctx)
 {
+       int ret;
+
+       assert(ua_ctx);
+
        if (ua_ctx->obj) {
-               ustctl_release_object(sock, ua_ctx->obj);
+               ret = ustctl_release_object(sock, ua_ctx->obj);
+               if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("UST app sock %d release context obj failed with ret %d",
+                                       sock, ret);
+               }
                free(ua_ctx->obj);
        }
        free(ua_ctx);
@@ -141,10 +191,18 @@ void delete_ust_app_ctx(int sock, struct ust_app_ctx *ua_ctx)
 static
 void delete_ust_app_event(int sock, struct ust_app_event *ua_event)
 {
+       int ret;
+
+       assert(ua_event);
+
        free(ua_event->filter);
 
        if (ua_event->obj != NULL) {
-               ustctl_release_object(sock, ua_event->obj);
+               ret = ustctl_release_object(sock, ua_event->obj);
+               if (ret < 0 && ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("UST app sock %d release event obj failed with ret %d",
+                                       sock, ret);
+               }
                free(ua_event->obj);
        }
        free(ua_event);
@@ -157,8 +215,16 @@ void delete_ust_app_event(int sock, struct ust_app_event *ua_event)
 static
 void delete_ust_app_stream(int sock, struct ust_app_stream *stream)
 {
+       int ret;
+
+       assert(stream);
+
        if (stream->obj) {
-               ustctl_release_object(sock, stream->obj);
+               ret = ustctl_release_object(sock, stream->obj);
+               if (ret < 0 && ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("UST app sock %d release stream obj failed with ret %d",
+                                       sock, ret);
+               }
                lttng_fd_put(LTTNG_FD_APPS, 2);
                free(stream->obj);
        }
@@ -178,6 +244,10 @@ void delete_ust_app_channel(int sock, struct ust_app_channel *ua_chan)
        struct ust_app_ctx *ua_ctx;
        struct ust_app_stream *stream, *stmp;
 
+       assert(ua_chan);
+
+       DBG3("UST app deleting channel %s", ua_chan->name);
+
        /* Wipe stream */
        cds_list_for_each_entry_safe(stream, stmp, &ua_chan->streams.head, list) {
                cds_list_del(&stream->list);
@@ -202,7 +272,11 @@ void delete_ust_app_channel(int sock, struct ust_app_channel *ua_chan)
        lttng_ht_destroy(ua_chan->events);
 
        if (ua_chan->obj != NULL) {
-               ustctl_release_object(sock, ua_chan->obj);
+               ret = ustctl_release_object(sock, ua_chan->obj);
+               if (ret < 0 && ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("UST app sock %d release channel obj failed with ret %d",
+                                       sock, ret);
+               }
                lttng_fd_put(LTTNG_FD_APPS, 2);
                free(ua_chan->obj);
        }
@@ -221,17 +295,7 @@ void delete_ust_app_session(int sock, struct ust_app_session *ua_sess)
        struct ust_app_channel *ua_chan;
 
        if (ua_sess->metadata) {
-               if (ua_sess->metadata->stream_obj) {
-                       ustctl_release_object(sock, ua_sess->metadata->stream_obj);
-                       lttng_fd_put(LTTNG_FD_APPS, 2);
-                       free(ua_sess->metadata->stream_obj);
-               }
-               if (ua_sess->metadata->obj) {
-                       ustctl_release_object(sock, ua_sess->metadata->obj);
-                       lttng_fd_put(LTTNG_FD_APPS, 2);
-                       free(ua_sess->metadata->obj);
-               }
-               trace_ust_destroy_metadata(ua_sess->metadata);
+               delete_ust_app_channel(sock, ua_sess->metadata);
        }
 
        cds_lfht_for_each_entry(ua_sess->channels->ht, &iter.iter, ua_chan,
@@ -243,7 +307,11 @@ void delete_ust_app_session(int sock, struct ust_app_session *ua_sess)
        lttng_ht_destroy(ua_sess->channels);
 
        if (ua_sess->handle != -1) {
-               ustctl_release_handle(sock, ua_sess->handle);
+               ret = ustctl_release_handle(sock, ua_sess->handle);
+               if (ret < 0 && ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("UST app sock %d release session handle failed with ret %d",
+                                       sock, ret);
+               }
        }
        free(ua_sess);
 }
@@ -313,6 +381,33 @@ void delete_ust_app_rcu(struct rcu_head *head)
        delete_ust_app(app);
 }
 
+/*
+ * Delete the session from the application ht and delete the data structure by
+ * freeing every object inside and releasing them.
+ */
+static void destroy_session(struct ust_app *app,
+               struct ust_app_session *ua_sess)
+{
+       int ret;
+       struct lttng_ht_iter iter;
+
+       assert(app);
+       assert(ua_sess);
+
+       iter.iter.node = &ua_sess->node.node;
+       ret = lttng_ht_del(app->sessions, &iter);
+       if (ret) {
+               /* Already scheduled for teardown. */
+               goto end;
+       }
+
+       /* Once deleted, free the data structure. */
+       delete_ust_app_session(app->sock, ua_sess);
+
+end:
+       return;
+}
+
 /*
  * Alloc new UST app session.
  */
@@ -325,15 +420,22 @@ struct ust_app_session *alloc_ust_app_session(void)
        ua_sess = zmalloc(sizeof(struct ust_app_session));
        if (ua_sess == NULL) {
                PERROR("malloc");
-               goto error;
+               goto error_free;
        }
 
        ua_sess->handle = -1;
        ua_sess->channels = lttng_ht_new(0, LTTNG_HT_TYPE_STRING);
 
+       if ((lttng_uuid_generate(ua_sess->uuid))) {
+               ERR("Failed to generate UST uuid");
+               goto error;
+       }
+
        return ua_sess;
 
 error:
+       free(ua_sess);
+error_free:
        return NULL;
 }
 
@@ -342,7 +444,7 @@ error:
  */
 static
 struct ust_app_channel *alloc_ust_app_channel(char *name,
-               struct lttng_ust_channel *attr)
+               struct lttng_ust_channel_attr *attr)
 {
        struct ust_app_channel *ua_chan;
 
@@ -359,6 +461,7 @@ struct ust_app_channel *alloc_ust_app_channel(char *name,
 
        ua_chan->enabled = 1;
        ua_chan->handle = -1;
+       ua_chan->key = get_next_channel_key();
        ua_chan->ctx = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
        ua_chan->events = lttng_ht_new(0, LTTNG_HT_TYPE_STRING);
        lttng_ht_node_init_str(&ua_chan->node, ua_chan->name);
@@ -367,7 +470,7 @@ struct ust_app_channel *alloc_ust_app_channel(char *name,
 
        /* Copy attributes */
        if (attr) {
-               /* Translate from lttng_ust_channel to lttng_ust_channel_attr.*/
+               /* Translate from lttng_ust_channel to ustctl_consumer_channel_attr. */
                ua_chan->attr.subbuf_size = attr->subbuf_size;
                ua_chan->attr.num_subbuf = attr->num_subbuf;
                ua_chan->attr.overwrite = attr->overwrite;
@@ -375,6 +478,8 @@ struct ust_app_channel *alloc_ust_app_channel(char *name,
                ua_chan->attr.read_timer_interval = attr->read_timer_interval;
                ua_chan->attr.output = attr->output;
        }
+       /* By default, the channel is a per cpu channel. */
+       ua_chan->attr.type = LTTNG_UST_CHAN_PER_CPU;
 
        DBG3("UST app channel %s allocated", ua_chan->name);
 
@@ -389,7 +494,7 @@ error:
  *
  * Return newly allocated stream pointer or NULL on error.
  */
-static struct ust_app_stream *alloc_ust_app_stream(void)
+struct ust_app_stream *ust_app_alloc_stream(void)
 {
        struct ust_app_stream *stream = NULL;
 
@@ -559,6 +664,12 @@ int create_ust_channel_context(struct ust_app_channel *ua_chan,
        ret = ustctl_add_context(app->sock, &ua_ctx->ctx,
                        ua_chan->obj, &ua_ctx->obj);
        if (ret < 0) {
+               if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("UST app create channel context failed for app (pid: %d) "
+                                       "with ret %d", app->pid, ret);
+               } else {
+                       DBG3("UST app disable event failed. Application is dead.");
+               }
                goto error;
        }
 
@@ -590,6 +701,12 @@ int set_ust_event_filter(struct ust_app_event *ua_event,
        ret = ustctl_set_filter(app->sock, ua_event->filter,
                        ua_event->obj);
        if (ret < 0) {
+               if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("UST app event %s filter failed for app (pid: %d) "
+                                       "with ret %d", ua_event->attr.name, app->pid, ret);
+               } else {
+                       DBG3("UST app filter event failed. Application is dead.");
+               }
                goto error;
        }
 
@@ -612,9 +729,13 @@ static int disable_ust_event(struct ust_app *app,
 
        ret = ustctl_disable(app->sock, ua_event->obj);
        if (ret < 0) {
-               ERR("UST app event %s disable failed for app (pid: %d) "
-                               "and session handle %d with ret %d",
-                               ua_event->attr.name, app->pid, ua_sess->handle, ret);
+               if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("UST app event %s disable failed for app (pid: %d) "
+                                       "and session handle %d with ret %d",
+                                       ua_event->attr.name, app->pid, ua_sess->handle, ret);
+               } else {
+                       DBG3("UST app disable event failed. Application is dead.");
+               }
                goto error;
        }
 
@@ -638,9 +759,13 @@ static int disable_ust_channel(struct ust_app *app,
 
        ret = ustctl_disable(app->sock, ua_chan->obj);
        if (ret < 0) {
-               ERR("UST app channel %s disable failed for app (pid: %d) "
-                               "and session handle %d with ret %d",
-                               ua_chan->name, app->pid, ua_sess->handle, ret);
+               if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("UST app channel %s disable failed for app (pid: %d) "
+                                       "and session handle %d with ret %d",
+                                       ua_chan->name, app->pid, ua_sess->handle, ret);
+               } else {
+                       DBG3("UST app disable channel failed. Application is dead.");
+               }
                goto error;
        }
 
@@ -664,9 +789,13 @@ static int enable_ust_channel(struct ust_app *app,
 
        ret = ustctl_enable(app->sock, ua_chan->obj);
        if (ret < 0) {
-               ERR("UST app channel %s enable failed for app (pid: %d) "
-                               "and session handle %d with ret %d",
-                               ua_chan->name, app->pid, ua_sess->handle, ret);
+               if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("UST app channel %s enable failed for app (pid: %d) "
+                                       "and session handle %d with ret %d",
+                                       ua_chan->name, app->pid, ua_sess->handle, ret);
+               } else {
+                       DBG3("UST app enable channel failed. Application is dead.");
+               }
                goto error;
        }
 
@@ -692,9 +821,13 @@ static int enable_ust_event(struct ust_app *app,
 
        ret = ustctl_enable(app->sock, ua_event->obj);
        if (ret < 0) {
-               ERR("UST app event %s enable failed for app (pid: %d) "
-                               "and session handle %d with ret %d",
-                               ua_event->attr.name, app->pid, ua_sess->handle, ret);
+               if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("UST app event %s enable failed for app (pid: %d) "
+                                       "and session handle %d with ret %d",
+                                       ua_event->attr.name, app->pid, ua_sess->handle, ret);
+               } else {
+                       DBG3("UST app enable event failed. Application is dead.");
+               }
                goto error;
        }
 
@@ -707,170 +840,87 @@ error:
 }
 
 /*
- * Open metadata onto the UST tracer for a UST session.
- */
-static int open_ust_metadata(struct ust_app *app,
-               struct ust_app_session *ua_sess)
-{
-       int ret;
-       struct lttng_ust_channel_attr uattr;
-
-       health_code_update();
-
-       uattr.overwrite = ua_sess->metadata->attr.overwrite;
-       uattr.subbuf_size = ua_sess->metadata->attr.subbuf_size;
-       uattr.num_subbuf = ua_sess->metadata->attr.num_subbuf;
-       uattr.switch_timer_interval =
-               ua_sess->metadata->attr.switch_timer_interval;
-       uattr.read_timer_interval =
-               ua_sess->metadata->attr.read_timer_interval;
-       uattr.output = ua_sess->metadata->attr.output;
-
-       /* We are going to receive 2 fds, we need to reserve them. */
-       ret = lttng_fd_get(LTTNG_FD_APPS, 2);
-       if (ret < 0) {
-               ERR("Exhausted number of available FD upon metadata open");
-               goto error;
-       }
-       /* UST tracer metadata creation */
-       ret = ustctl_open_metadata(app->sock, ua_sess->handle, &uattr,
-                       &ua_sess->metadata->obj);
-       if (ret < 0) {
-               ERR("UST app open metadata failed for app pid:%d with ret %d",
-                               app->pid, ret);
-               goto error;
-       }
-
-       ua_sess->metadata->handle = ua_sess->metadata->obj->handle;
-
-error:
-       health_code_update();
-       return ret;
-}
-
-/*
- * Create metadata stream onto the UST tracer for a given session.
- */
-static int create_ust_metadata_stream(struct ust_app *app,
-               struct ust_app_session *ua_sess)
-{
-       int ret;
-
-       health_code_update();
-
-       /* We are going to receive 2 fds, we need to reserve them. */
-       ret = lttng_fd_get(LTTNG_FD_APPS, 2);
-       if (ret < 0) {
-               ERR("Exhausted number of available FD upon metadata stream create");
-               goto error;
-       }
-       ret = ustctl_create_stream(app->sock, ua_sess->metadata->obj,
-                       &ua_sess->metadata->stream_obj);
-       if (ret < 0) {
-               lttng_fd_put(LTTNG_FD_APPS, 2);
-               ERR("UST create metadata stream failed");
-               goto error;
-       }
-
-error:
-       health_code_update();
-       return ret;
-}
-
-/*
- * Create stream onto the UST tracer for a given channel.
+ * Create the specified channel onto the UST tracer for a UST session.
  *
- * Return -ENOENT if no more stream is available for this channel.
- * On success, return 0.
- * On error, return a negative value.
+ * Return 0 on success. On error, a negative value is returned.
  */
-static int create_ust_stream(struct ust_app *app,
-               struct ust_app_channel *ua_chan, struct ust_app_stream *stream)
+static int create_ust_channel(struct ust_app *app,
+               struct ust_app_session *ua_sess, struct ust_app_channel *ua_chan,
+               struct consumer_output *consumer)
 {
        int ret;
+       unsigned int nb_fd = 0;
+       struct consumer_socket *socket;
+       struct ust_app_stream *stream, *stmp;
 
        assert(app);
+       assert(ua_sess);
        assert(ua_chan);
-       assert(ua_chan->obj);
-       assert(stream);
+       assert(consumer);
 
        health_code_update();
 
-       /* We are going to receive 2 fds, we need to reserve them. */
-       ret = lttng_fd_get(LTTNG_FD_APPS, 2);
-       if (ret < 0) {
-               ERR("Exhausted number of available FD on stream creation");
-               /* Just to make sure we never return -ENOENT. */
+       /* Get the right consumer socket for the application. */
+       socket = find_consumer_socket_by_bitness(app->bits_per_long, consumer);
+       if (!socket) {
                ret = -1;
                goto error;
        }
 
+       health_code_update();
+
        /*
-        * Set the stream name before creating it. On error, we don't have to
-        * delete it on the tracer side.
+        * Ask consumer to create channel. The consumer will return the number of
+        * stream we have to expect.
         */
-       ret = snprintf(stream->name, sizeof(stream->name), "%s_%u",
-                       ua_chan->name, ua_chan->streams.count);
+       ret = ust_consumer_ask_channel(ua_sess, ua_chan, consumer, socket);
        if (ret < 0) {
-               /* Without the stream name we can't continue using it. */
-               PERROR("snprintf UST create stream");
-               /* Just to make sure we never return -ENOENT. */
-               ret = -1;
                goto error;
        }
 
-       ret = ustctl_create_stream(app->sock, ua_chan->obj, &stream->obj);
+       /*
+        * Compute the number of fd needed before receiving them. It must be 2 per
+        * stream.
+        */
+       nb_fd = DEFAULT_UST_STREAM_FD_NUM * ua_chan->expected_stream_count;
+
+       /* Reserve the amount of file descriptor we need. */
+       ret = lttng_fd_get(LTTNG_FD_APPS, nb_fd);
        if (ret < 0) {
-               lttng_fd_put(LTTNG_FD_APPS, 2);
-               /* Indicates that there is no more stream for that channel. */
-               if (ret != -LTTNG_UST_ERR_NOENT) {
-                       ERR("UST create metadata stream failed (ret: %d)", ret);
-               }
-               goto error;
+               ERR("Exhausted number of available FD upon create channel");
+               goto error_fd_get;
        }
 
-       /* Set stream handle with the returned value. */
-       stream->handle = stream->obj->handle;
-
-error:
        health_code_update();
-       return ret;
-}
 
-/*
- * Create the specified channel onto the UST tracer for a UST session.
- */
-static int create_ust_channel(struct ust_app *app,
-               struct ust_app_session *ua_sess, struct ust_app_channel *ua_chan)
-{
-       int ret;
-
-       health_code_update();
-
-       /* We are going to receive 2 fds, we need to reserve them. */
-       ret = lttng_fd_get(LTTNG_FD_APPS, 2);
+       /*
+        * Now get the channel from the consumer. This call wil populate the stream
+        * list of that channel and set the ust object.
+        */
+       ret = ust_consumer_get_channel(socket, ua_chan);
        if (ret < 0) {
-               ERR("Exhausted number of available FD upon create channel");
-               goto error;
+               goto error_destroy;
        }
 
-       health_code_update();
-
-       ret = ustctl_create_channel(app->sock, ua_sess->handle, &ua_chan->attr,
-                       &ua_chan->obj);
+       /* Send channel to the application. */
+       ret = ust_consumer_send_channel_to_ust(app, ua_sess, ua_chan);
        if (ret < 0) {
-               ERR("Creating channel %s for app (pid: %d, sock: %d) "
-                               "and session handle %d with ret %d",
-                               ua_chan->name, app->pid, app->sock,
-                               ua_sess->handle, ret);
-               lttng_fd_put(LTTNG_FD_APPS, 2);
                goto error;
        }
 
-       ua_chan->handle = ua_chan->obj->handle;
+       /* Send all streams to application. */
+       cds_list_for_each_entry_safe(stream, stmp, &ua_chan->streams.head, list) {
+               ret = ust_consumer_send_stream_to_ust(app, ua_chan, stream);
+               if (ret < 0) {
+                       goto error;
+               }
+               /* We don't need the stream anymore once sent to the tracer. */
+               cds_list_del(&stream->list);
+               delete_ust_app_stream(-1, stream);
+       }
 
-       DBG2("UST app channel %s created successfully for pid:%d and sock:%d",
-                       ua_chan->name, app->pid, app->sock);
+       /* Flag the channel that it is sent to the application. */
+       ua_chan->is_sent = 1;
 
        health_code_update();
 
@@ -882,6 +932,18 @@ static int create_ust_channel(struct ust_app *app,
                }
        }
 
+       return 0;
+
+error_destroy:
+       lttng_fd_put(LTTNG_FD_APPS, nb_fd);
+error_fd_get:
+       /*
+        * Initiate a destroy channel on the consumer since we had an error
+        * handling it on our side. The return value is of no importance since we
+        * already have a ret value set by the previous error that we need to
+        * return.
+        */
+       (void) ust_consumer_destroy_channel(socket, ua_chan);
 error:
        health_code_update();
        return ret;
@@ -902,8 +964,12 @@ int create_ust_event(struct ust_app *app, struct ust_app_session *ua_sess,
        ret = ustctl_create_event(app->sock, &ua_event->attr, ua_chan->obj,
                        &ua_event->obj);
        if (ret < 0) {
-               ERR("Error ustctl create event %s for app pid: %d with ret %d",
-                               ua_event->attr.name, app->pid, ret);
+               if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("Error ustctl create event %s for app pid: %d with ret %d",
+                                       ua_event->attr.name, app->pid, ret);
+               } else {
+                       DBG3("UST app create event failed. Application is dead.");
+               }
                goto error;
        }
 
@@ -988,8 +1054,18 @@ static void shadow_copy_channel(struct ust_app_channel *ua_chan,
 
        strncpy(ua_chan->name, uchan->name, sizeof(ua_chan->name));
        ua_chan->name[sizeof(ua_chan->name) - 1] = '\0';
-       /* Copy event attributes */
-       memcpy(&ua_chan->attr, &uchan->attr, sizeof(ua_chan->attr));
+
+       /* Copy event attributes since the layout is different. */
+       ua_chan->attr.subbuf_size = uchan->attr.subbuf_size;
+       ua_chan->attr.num_subbuf = uchan->attr.num_subbuf;
+       ua_chan->attr.overwrite = uchan->attr.overwrite;
+       ua_chan->attr.switch_timer_interval = uchan->attr.switch_timer_interval;
+       ua_chan->attr.read_timer_interval = uchan->attr.read_timer_interval;
+       ua_chan->attr.output = uchan->attr.output;
+       /*
+        * Note that the attribute channel type is not set since the channel on the
+        * tracing registry side does not have this information.
+        */
 
        ua_chan->enabled = uchan->enabled;
 
@@ -1077,8 +1153,14 @@ static void shadow_copy_session(struct ust_app_session *ua_sess,
                        /* malloc failed FIXME: Might want to do handle ENOMEM .. */
                        continue;
                }
-
                shadow_copy_channel(ua_chan, uchan);
+               /*
+                * The concept of metadata channel does not exist on the tracing
+                * registry side of the session daemon so this can only be a per CPU
+                * channel and not metadata.
+                */
+               ua_chan->attr.type = LTTNG_UST_CHAN_PER_CPU;
+
                lttng_ht_add_unique_str(ua_sess->channels, &ua_chan->node);
        }
 }
@@ -1159,7 +1241,12 @@ static int create_ust_app_session(struct ltt_ust_session *usess,
        if (ua_sess->handle == -1) {
                ret = ustctl_create_session(app->sock);
                if (ret < 0) {
-                       ERR("Creating session for app pid %d", app->pid);
+                       if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                               ERR("Creating session for app pid %d with ret %d",
+                                               app->pid, ret);
+                       } else {
+                               DBG("UST app creating session failed. Application is dead");
+                       }
                        delete_ust_app_session(-1, ua_sess);
                        if (ret != -ENOMEM) {
                                /*
@@ -1328,6 +1415,7 @@ error:
  */
 static int create_ust_app_channel(struct ust_app_session *ua_sess,
                struct ltt_ust_channel *uchan, struct ust_app *app,
+               struct consumer_output *consumer, enum lttng_ust_chan_type type,
                struct ust_app_channel **ua_chanp)
 {
        int ret = 0;
@@ -1351,10 +1439,11 @@ static int create_ust_app_channel(struct ust_app_session *ua_sess,
        }
        shadow_copy_channel(ua_chan, uchan);
 
-       ret = create_ust_channel(app, ua_sess, ua_chan);
+       /* Set channel type. */
+       ua_chan->attr.type = type;
+
+       ret = create_ust_channel(app, ua_sess, ua_chan, consumer);
        if (ret < 0) {
-               /* Not found previously means that it does not exist on the tracer */
-               assert(ret != -LTTNG_UST_ERR_EXIST);
                goto error;
        }
 
@@ -1373,7 +1462,7 @@ end:
        return 0;
 
 error:
-       delete_ust_app_channel(-1, ua_chan);
+       delete_ust_app_channel(ua_chan->is_sent ? app->sock : -1, ua_chan);
        return ret;
 }
 
@@ -1431,60 +1520,51 @@ error:
  * Create UST metadata and open it on the tracer side.
  */
 static int create_ust_app_metadata(struct ust_app_session *ua_sess,
-               char *pathname, struct ust_app *app)
+               struct ust_app *app, struct consumer_output *consumer)
 {
        int ret = 0;
+       struct ust_app_channel *metadata;
 
-       if (ua_sess->metadata == NULL) {
-               /* Allocate UST metadata */
-               ua_sess->metadata = trace_ust_create_metadata(pathname);
-               if (ua_sess->metadata == NULL) {
-                       /* malloc() failed */
-                       goto error;
-               }
-
-               ret = open_ust_metadata(app, ua_sess);
-               if (ret < 0) {
-                       DBG3("Opening metadata failed. Cleaning up memory");
-
-                       /* Cleanup failed metadata struct */
-                       free(ua_sess->metadata);
-                       /*
-                        * This is very important because delete_ust_app_session check if
-                        * the pointer is null or not in order to delete the metadata.
-                        */
-                       ua_sess->metadata = NULL;
-                       goto error;
-               }
+       assert(ua_sess);
+       assert(app);
 
-               DBG2("UST metadata opened for app pid %d", app->pid);
+       if (ua_sess->metadata) {
+               /* Already exist. Return success. */
+               goto end;
        }
 
-       /* Open UST metadata stream */
-       if (ua_sess->metadata->stream_obj == NULL) {
-               ret = create_ust_metadata_stream(app, ua_sess);
-               if (ret < 0) {
-                       goto error;
-               }
+       /* Allocate UST metadata */
+       metadata = alloc_ust_app_channel(DEFAULT_METADATA_NAME, NULL);
+       if (!metadata) {
+               /* malloc() failed */
+               ret = -ENOMEM;
+               goto error;
+       }
 
-               ret = snprintf(ua_sess->metadata->pathname, PATH_MAX,
-                               "%s/" DEFAULT_METADATA_NAME, ua_sess->path);
-               if (ret < 0) {
-                       PERROR("asprintf UST create stream");
-                       goto error;
-               }
+       /* Set default attributes for metadata. */
+       metadata->attr.overwrite = DEFAULT_CHANNEL_OVERWRITE;
+       metadata->attr.subbuf_size = default_get_metadata_subbuf_size();
+       metadata->attr.num_subbuf = DEFAULT_METADATA_SUBBUF_NUM;
+       metadata->attr.switch_timer_interval = DEFAULT_CHANNEL_SWITCH_TIMER;
+       metadata->attr.read_timer_interval = DEFAULT_CHANNEL_READ_TIMER;
+       metadata->attr.output = LTTNG_UST_MMAP;
+       metadata->attr.type = LTTNG_UST_CHAN_METADATA;
 
-               DBG2("UST metadata stream object created for app pid %d",
-                               app->pid);
-       } else {
-               ERR("Attempting to create stream without metadata opened");
-               goto error;
+       ret = create_ust_channel(app, ua_sess, metadata, consumer);
+       if (ret < 0) {
+               goto error_create;
        }
 
-       return 0;
+       ua_sess->metadata = metadata;
+
+       DBG2("UST metadata opened for app pid %d", app->pid);
 
+end:
+       return 0;
+error_create:
+       delete_ust_app_channel(metadata->is_sent ? app->sock : -1, metadata);
 error:
-       return -1;
+       return ret;
 }
 
 /*
@@ -1725,13 +1805,27 @@ int ust_app_list_events(struct lttng_event **events)
                }
                handle = ustctl_tracepoint_list(app->sock);
                if (handle < 0) {
-                       ERR("UST app list events getting handle failed for app pid %d",
-                                       app->pid);
+                       if (handle != -EPIPE && handle != -LTTNG_UST_ERR_EXITING) {
+                               ERR("UST app list events getting handle failed for app pid %d",
+                                               app->pid);
+                       }
                        continue;
                }
 
                while ((ret = ustctl_tracepoint_list_get(app->sock, handle,
                                        &uiter)) != -LTTNG_UST_ERR_NOENT) {
+                       /* Handle ustctl error. */
+                       if (ret < 0) {
+                               free(tmp_event);
+                               if (ret != -LTTNG_UST_ERR_EXITING || ret != -EPIPE) {
+                                       ERR("UST app tp list get failed for app %d with ret %d",
+                                                       app->sock, ret);
+                               } else {
+                                       DBG3("UST app tp list get failed. Application is dead");
+                               }
+                               goto rcu_error;
+                       }
+
                        health_code_update();
                        if (count >= nbmem) {
                                /* In case the realloc fails, we free the memory */
@@ -1805,13 +1899,27 @@ int ust_app_list_event_fields(struct lttng_event_field **fields)
                }
                handle = ustctl_tracepoint_field_list(app->sock);
                if (handle < 0) {
-                       ERR("UST app list event fields getting handle failed for app pid %d",
-                                       app->pid);
+                       if (handle != -EPIPE && handle != -LTTNG_UST_ERR_EXITING) {
+                               ERR("UST app list field getting handle failed for app pid %d",
+                                               app->pid);
+                       }
                        continue;
                }
 
                while ((ret = ustctl_tracepoint_field_list_get(app->sock, handle,
                                        &uiter)) != -LTTNG_UST_ERR_NOENT) {
+                       /* Handle ustctl error. */
+                       if (ret < 0) {
+                               free(tmp_event);
+                               if (ret != -LTTNG_UST_ERR_EXITING || ret != -EPIPE) {
+                                       ERR("UST app tp list field failed for app %d with ret %d",
+                                                       app->sock, ret);
+                               } else {
+                                       DBG3("UST app tp list field failed. Application is dead");
+                               }
+                               goto rcu_error;
+                       }
+
                        health_code_update();
                        if (count >= nbmem) {
                                /* In case the realloc fails, we free the memory */
@@ -2186,7 +2294,8 @@ int ust_app_create_channel_glb(struct ltt_ust_session *usess,
                assert(ua_sess);
 
                /* Create channel onto application. We don't need the chan ref. */
-               ret = create_ust_app_channel(ua_sess, uchan, app, NULL);
+               ret = create_ust_app_channel(ua_sess, uchan, app, usess->consumer,
+                               LTTNG_UST_CHAN_PER_CPU, NULL);
                if (ret < 0) {
                        if (ret == -ENOMEM) {
                                /* No more memory is a fatal error. Stop right now. */
@@ -2194,7 +2303,7 @@ int ust_app_create_channel_glb(struct ltt_ust_session *usess,
                        }
                        /* Cleanup the created session if it's the case. */
                        if (created) {
-                               delete_ust_app_session(app->sock, ua_sess);
+                               destroy_session(app, ua_sess);
                        }
                }
        }
@@ -2337,11 +2446,7 @@ int ust_app_create_event_glb(struct ltt_ust_session *usess,
 int ust_app_start_trace(struct ltt_ust_session *usess, struct ust_app *app)
 {
        int ret = 0;
-       struct lttng_ht_iter iter;
        struct ust_app_session *ua_sess;
-       struct ust_app_channel *ua_chan;
-       struct ust_app_stream *ustream;
-       struct consumer_socket *socket;
 
        DBG("Starting tracing for ust app pid %d", app->pid);
 
@@ -2370,79 +2475,13 @@ int ust_app_start_trace(struct ltt_ust_session *usess, struct ust_app *app)
                if (ret < 0) {
                        if (ret != -EEXIST) {
                                ERR("Trace directory creation error");
-                               ret = -1;
-                               goto error_rcu_unlock;
-                       }
-               }
-       }
-
-       ret = create_ust_app_metadata(ua_sess, usess->pathname, app);
-       if (ret < 0) {
-               ret = LTTNG_ERR_UST_META_FAIL;
-               goto error_rcu_unlock;
-       }
-
-       /* For each channel */
-       cds_lfht_for_each_entry(ua_sess->channels->ht, &iter.iter, ua_chan,
-                       node.node) {
-               /* Create all streams */
-               while (1) {
-                       /* Create UST stream */
-                       ustream = alloc_ust_app_stream();
-                       if (ustream == NULL) {
                                goto error_rcu_unlock;
                        }
-
-                       health_code_update();
-
-                       ret = create_ust_stream(app, ua_chan, ustream);
-                       if (ret < 0) {
-                               /* Free unused memory after this point. */
-                               free(ustream);
-                               if (ret == -LTTNG_UST_ERR_NOENT) {
-                                       /* Got all streams. Continue normal execution. */
-                                       break;
-                               }
-                               /* Error at this point. Stop everything. */
-                               ret = LTTNG_ERR_UST_STREAM_FAIL;
-                               goto error_rcu_unlock;
-                       }
-
-                       health_code_update();
-
-                       /* Order is important this is why a list is used. */
-                       cds_list_add_tail(&ustream->list, &ua_chan->streams.head);
-                       ua_chan->streams.count++;
-
-                       DBG2("UST stream %d ready (handle: %d)", ua_chan->streams.count,
-                                       ustream->handle);
-               }
-
-               health_code_update();
-       }
-
-       switch (app->bits_per_long) {
-       case 64:
-               socket = consumer_find_socket(uatomic_read(&ust_consumerd64_fd),
-                               usess->consumer);
-               if (socket == NULL) {
-                       goto skip_setup;
-               }
-               break;
-       case 32:
-               socket = consumer_find_socket(uatomic_read(&ust_consumerd32_fd),
-                               usess->consumer);
-               if (socket == NULL) {
-                       goto skip_setup;
                }
-               break;
-       default:
-               ret = -EINVAL;
-               goto error_rcu_unlock;
        }
 
-       /* Setup UST consumer socket and send fds to it */
-       ret = ust_consumer_send_session(ua_sess, usess->consumer, socket);
+       /* Create the metadata for the application. */
+       ret = create_ust_app_metadata(ua_sess, app, usess->consumer);
        if (ret < 0) {
                goto error_rcu_unlock;
        }
@@ -2453,7 +2492,12 @@ skip_setup:
        /* This start the UST tracing */
        ret = ustctl_start_session(app->sock, ua_sess->handle);
        if (ret < 0) {
-               ERR("Error starting tracing for app pid: %d (ret: %d)", app->pid, ret);
+               if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("Error starting tracing for app pid: %d (ret: %d)",
+                                       app->pid, ret);
+               } else {
+                       DBG("UST app start session failed. Application is dead.");
+               }
                goto error_rcu_unlock;
        }
 
@@ -2463,7 +2507,11 @@ skip_setup:
        health_code_update();
 
        /* Quiescent wait after starting trace */
-       ustctl_wait_quiescent(app->sock);
+       ret = ustctl_wait_quiescent(app->sock);
+       if (ret < 0 && ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+               ERR("UST app wait quiescent failed for app pid %d ret %d",
+                               app->pid, ret);
+       }
 
 end:
        rcu_read_unlock();
@@ -2514,14 +2562,23 @@ int ust_app_stop_trace(struct ltt_ust_session *usess, struct ust_app *app)
        /* This inhibits UST tracing */
        ret = ustctl_stop_session(app->sock, ua_sess->handle);
        if (ret < 0) {
-               ERR("Error stopping tracing for app pid: %d (ret: %d)", app->pid, ret);
+               if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("Error stopping tracing for app pid: %d (ret: %d)",
+                                       app->pid, ret);
+               } else {
+                       DBG("UST app stop session failed. Application is dead.");
+               }
                goto error_rcu_unlock;
        }
 
        health_code_update();
 
        /* Quiescent wait after stopping trace */
-       ustctl_wait_quiescent(app->sock);
+       ret = ustctl_wait_quiescent(app->sock);
+       if (ret < 0 && ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+               ERR("UST app wait quiescent failed for app pid %d ret %d",
+                               app->pid, ret);
+       }
 
        health_code_update();
 
@@ -2529,10 +2586,18 @@ int ust_app_stop_trace(struct ltt_ust_session *usess, struct ust_app *app)
        cds_lfht_for_each_entry(ua_sess->channels->ht, &iter.iter, ua_chan,
                        node.node) {
                health_code_update();
+               assert(ua_chan->is_sent);
                ret = ustctl_sock_flush_buffer(app->sock, ua_chan->obj);
                if (ret < 0) {
-                       ERR("UST app PID %d channel %s flush failed with ret %d",
-                                       app->pid, ua_chan->name, ret);
+                       if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                               ERR("UST app PID %d channel %s flush failed with ret %d",
+                                               app->pid, ua_chan->name, ret);
+                       } else {
+                               DBG3("UST app failed to flush %s. Application is dead.",
+                                               ua_chan->name);
+                               /* No need to continue. */
+                               goto end;
+                       }
                        /* Continuing flushing all buffers */
                        continue;
                }
@@ -2540,11 +2605,17 @@ int ust_app_stop_trace(struct ltt_ust_session *usess, struct ust_app *app)
 
        health_code_update();
 
+       assert(ua_sess->metadata->is_sent);
        /* Flush all buffers before stopping */
        ret = ustctl_sock_flush_buffer(app->sock, ua_sess->metadata->obj);
        if (ret < 0) {
-               ERR("UST app PID %d metadata flush failed with ret %d", app->pid,
-                               ret);
+               if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("UST app PID %d metadata flush failed with ret %d", app->pid,
+                                       ret);
+                       goto error_rcu_unlock;
+               } else {
+                       DBG3("UST app failed to flush metadata. Application is dead.");
+               }
        }
 
 end:
@@ -2563,11 +2634,10 @@ error_rcu_unlock:
  */
 static int destroy_trace(struct ltt_ust_session *usess, struct ust_app *app)
 {
+       int ret;
        struct ust_app_session *ua_sess;
-       struct lttng_ust_object_data obj;
        struct lttng_ht_iter iter;
        struct lttng_ht_node_ulong *node;
-       int ret;
 
        DBG("Destroy tracing for ust app pid %d", app->pid);
 
@@ -2584,24 +2654,18 @@ static int destroy_trace(struct ltt_ust_session *usess, struct ust_app *app)
                goto end;
        }
        ua_sess = caa_container_of(node, struct ust_app_session, node);
-       ret = lttng_ht_del(app->sessions, &iter);
-       if (ret) {
-               /* Already scheduled for teardown. */
-               goto end;
-       }
 
-       obj.handle = ua_sess->handle;
-       obj.shm_fd = -1;
-       obj.wait_fd = -1;
-       obj.memory_map_size = 0;
        health_code_update();
-       ustctl_release_object(app->sock, &obj);
+       destroy_session(app, ua_sess);
 
        health_code_update();
-       delete_ust_app_session(app->sock, ua_sess);
 
        /* Quiescent wait after stopping trace */
-       ustctl_wait_quiescent(app->sock);
+       ret = ustctl_wait_quiescent(app->sock);
+       if (ret < 0 && ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+               ERR("UST app wait quiescent failed for app pid %d ret %d",
+                               app->pid, ret);
+       }
 
 end:
        rcu_read_unlock();
@@ -2701,6 +2765,7 @@ void ust_app_global_update(struct ltt_ust_session *usess, int sock)
        struct ust_app_ctx *ua_ctx;
 
        assert(usess);
+       assert(sock >= 0);
 
        DBG2("UST app global update for app sock %d for session id %d", sock,
                        usess->id);
@@ -2709,7 +2774,7 @@ void ust_app_global_update(struct ltt_ust_session *usess, int sock)
 
        app = find_app_by_sock(sock);
        if (app == NULL) {
-               ERR("Failed to update app sock %d", sock);
+               ERR("Failed to find app sock %d", sock);
                goto error;
        }
 
@@ -2731,18 +2796,21 @@ void ust_app_global_update(struct ltt_ust_session *usess, int sock)
         */
        cds_lfht_for_each_entry(ua_sess->channels->ht, &iter.iter, ua_chan,
                        node.node) {
-               ret = create_ust_channel(app, ua_sess, ua_chan);
+               ret = create_ust_channel(app, ua_sess, ua_chan, usess->consumer);
                if (ret < 0) {
-                       /* FIXME: Should we quit here or continue... */
-                       continue;
+                       /*
+                        * Stop everything. On error, the application failed, no more file
+                        * descriptor are available or ENOMEM so stopping here is the only
+                        * thing we can do for now.
+                        */
+                       goto error;
                }
 
                cds_lfht_for_each_entry(ua_chan->ctx->ht, &iter_ctx.iter, ua_ctx,
                                node.node) {
                        ret = create_ust_channel_context(ua_chan, ua_ctx, app);
                        if (ret < 0) {
-                               /* FIXME: Should we quit here or continue... */
-                               continue;
+                               goto error;
                        }
                }
 
@@ -2752,8 +2820,7 @@ void ust_app_global_update(struct ltt_ust_session *usess, int sock)
                                node.node) {
                        ret = create_ust_event(app, ua_sess, ua_chan, ua_event);
                        if (ret < 0) {
-                               /* FIXME: Should we quit here or continue... */
-                               continue;
+                               goto error;
                        }
                }
        }
@@ -2767,7 +2834,14 @@ void ust_app_global_update(struct ltt_ust_session *usess, int sock)
                DBG2("UST trace started for app pid %d", app->pid);
        }
 
+       /* Everything went well at this point. */
+       rcu_read_unlock();
+       return;
+
 error:
+       if (ua_sess) {
+               destroy_session(app, ua_sess);
+       }
        rcu_read_unlock();
        return;
 }
@@ -2962,6 +3036,9 @@ int ust_app_validate_version(int sock)
 
        ret = ustctl_tracer_version(sock, &app->version);
        if (ret < 0) {
+               if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("UST app tracer version failed for app pid %d", app->pid);
+               }
                goto error;
        }
 
@@ -3018,7 +3095,6 @@ int ust_app_calibrate_glb(struct lttng_ust_calibrate *calibrate)
                                ret = 0;
                                break;
                        default:
-                               /* TODO: Report error to user */
                                DBG2("Calibrate app PID %d returned with error %d",
                                                app->pid, ret);
                                break;
index 1d774a7b96c82ba8aac956020e1f136b28a6de03..e8bb9a980ab1cf8f122016405638c8735341cfdb 100644 (file)
@@ -20,6 +20,7 @@
 
 #include <stdint.h>
 
+#include <common/compat/uuid.h>
 #include "trace-ust.h"
 
 /* lttng-ust supported version. */
@@ -96,14 +97,22 @@ struct ust_app_stream {
        struct lttng_ust_object_data *obj;
        /* Using a list of streams to keep order. */
        struct cds_list_head list;
+       struct ustctl_consumer_stream *ustream;
 };
 
 struct ust_app_channel {
        int enabled;
        int handle;
+       /* Channel and streams were sent to the UST tracer. */
+       int is_sent;
+       /* Unique key used to identify the channel on the consumer side. */
+       unsigned long key;
+       /* Number of stream that this channel is expected to receive. */
+       unsigned int expected_stream_count;
        char name[LTTNG_UST_SYM_NAME_LEN];
-       struct lttng_ust_channel_attr attr;
        struct lttng_ust_object_data *obj;
+       struct ustctl_consumer_channel_attr attr;
+       struct ustctl_consumer_channel *channel;
        struct ust_app_stream_list streams;
        struct lttng_ht *ctx;
        struct lttng_ht *events;
@@ -116,7 +125,7 @@ struct ust_app_session {
        int started;  /* allows detection of start vs restart. */
        int handle;   /* used has unique identifier for app session */
        int id;       /* session unique identifier */
-       struct ltt_ust_metadata *metadata;
+       struct ust_app_channel *metadata;
        struct lttng_ht *channels; /* Registered channels */
        struct lttng_ht_node_ulong node;
        char path[PATH_MAX];
@@ -124,6 +133,8 @@ struct ust_app_session {
        uid_t uid;
        gid_t gid;
        struct cds_list_head teardown_node;
+       /* Universal unique identifier used by the tracer. */
+       unsigned char uuid[UUID_STR_LEN];
 };
 
 /*
@@ -208,6 +219,7 @@ struct lttng_ht *ust_app_get_ht(void);
 struct ust_app *ust_app_find_by_pid(pid_t pid);
 int ust_app_validate_version(int sock);
 int ust_app_calibrate_glb(struct lttng_ust_calibrate *calibrate);
+struct ust_app_stream *ust_app_alloc_stream(void);
 
 #else /* HAVE_LIBLTTNG_UST_CTL */
 
index f108e99bd0b1379bd751432bee5276394d97b054..7c1ae402c8ded33c75532d76530ba5ddf195001d 100644 (file)
 #include "ust-consumer.h"
 
 /*
- * Send a single channel to the consumer using command ADD_CHANNEL.
+ * Return allocated full pathname of the session using the consumer trace path
+ * and subdir if available. On a successful allocation, the directory of the
+ * trace is created with the session credentials.
+ *
+ * The caller can safely free(3) the returned value. On error, NULL is
+ * returned.
  */
-static int send_channel(struct consumer_socket *sock,
-               struct ust_app_channel *uchan)
+static char *setup_trace_path(struct consumer_output *consumer,
+               struct ust_app_session *ua_sess)
 {
-       int ret, fd;
-       struct lttcomm_consumer_msg msg;
-
-       /* Safety net */
-       assert(uchan);
-       assert(sock);
-
-       if (sock->fd < 0) {
-               ret = -EINVAL;
-               goto error;
-       }
-
-       DBG2("Sending channel %s to UST consumer", uchan->name);
+       int ret;
+       char *pathname;
 
-       consumer_init_channel_comm_msg(&msg,
-                       LTTNG_CONSUMER_ADD_CHANNEL,
-                       uchan->obj->shm_fd,
-                       uchan->attr.subbuf_size,
-                       uchan->obj->memory_map_size,
-                       uchan->name,
-                       uchan->streams.count);
+       assert(consumer);
+       assert(ua_sess);
 
        health_code_update();
 
-       ret = consumer_send_channel(sock, &msg);
-       if (ret < 0) {
+       /* Allocate our self the string to make sure we never exceed PATH_MAX. */
+       pathname = zmalloc(PATH_MAX);
+       if (!pathname) {
                goto error;
        }
 
-       health_code_update();
+       /* 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);
+               if (ret < 0) {
+                       PERROR("snprintf channel path");
+                       goto error;
+               }
 
-       fd = uchan->obj->shm_fd;
-       ret = consumer_send_fds(sock, &fd, 1);
-       if (ret < 0) {
-               goto error;
+               /* Create directory. Ignore if exist. */
+               ret = run_as_mkdir_recursive(pathname, S_IRWXU | S_IRWXG, ua_sess->uid,
+                               ua_sess->gid);
+               if (ret < 0) {
+                       if (ret != -EEXIST) {
+                               ERR("Trace directory creation error");
+                               goto error;
+                       }
+               }
+       } else {
+               ret = snprintf(pathname, PATH_MAX, "%s/%s", consumer->subdir,
+                               ua_sess->path);
+               if (ret < 0) {
+                       PERROR("snprintf channel path");
+                       goto error;
+               }
        }
 
-       health_code_update();
+       return pathname;
 
 error:
-       return ret;
+       free(pathname);
+       return NULL;
 }
 
 /*
- * Send a single stream to the consumer using ADD_STREAM command.
+ * Send a single channel to the consumer using command ADD_CHANNEL.
+ *
+ * Consumer socket MUST be acquired before calling this.
  */
-static int send_channel_stream(struct consumer_socket *sock,
-               struct ust_app_channel *uchan, struct ust_app_session *usess,
-               struct ust_app_stream *stream, struct consumer_output *consumer,
-               const char *pathname)
+static int ask_channel_creation(struct ust_app_session *ua_sess,
+               struct ust_app_channel *ua_chan, struct consumer_output *consumer,
+               struct consumer_socket *socket)
 {
-       int ret, fds[2];
+       int ret;
+       unsigned long key;
+       char *pathname = NULL;
        struct lttcomm_consumer_msg msg;
 
-       /* Safety net */
-       assert(uchan);
-       assert(usess);
-       assert(stream);
+       assert(ua_sess);
+       assert(ua_chan);
+       assert(socket);
        assert(consumer);
-       assert(sock);
-
-       DBG2("Sending stream %d of channel %s to kernel consumer",
-                       stream->obj->shm_fd, uchan->name);
-
-       consumer_init_stream_comm_msg(&msg,
-                       LTTNG_CONSUMER_ADD_STREAM,
-                       uchan->obj->shm_fd,
-                       stream->obj->shm_fd,
-                       LTTNG_CONSUMER_ACTIVE_STREAM,
-                       DEFAULT_UST_CHANNEL_OUTPUT,
-                       stream->obj->memory_map_size,
-                       usess->uid,
-                       usess->gid,
-                       consumer->net_seq_index,
-                       0, /* Metadata flag unset */
-                       stream->name,
+
+       DBG2("Asking UST consumer for channel");
+
+       /* Get and create full trace path of session. */
+       pathname = setup_trace_path(consumer, ua_sess);
+       if (!pathname) {
+               ret = -1;
+               goto error;
+       }
+
+       consumer_init_ask_channel_comm_msg(&msg,
+                       ua_chan->attr.subbuf_size,
+                       ua_chan->attr.num_subbuf,
+                       ua_chan->attr.overwrite,
+                       ua_chan->attr.switch_timer_interval,
+                       ua_chan->attr.read_timer_interval,
+                       (int) ua_chan->attr.output,
+                       (int) ua_chan->attr.type,
+                       ua_sess->id,
                        pathname,
-                       usess->id);
+                       ua_chan->name,
+                       ua_sess->uid,
+                       ua_sess->gid,
+                       consumer->net_seq_index,
+                       ua_chan->key,
+                       ua_sess->uuid);
 
        health_code_update();
 
-       /* Send stream and file descriptor */
-       fds[0] = stream->obj->shm_fd;
-       fds[1] = stream->obj->wait_fd;
-       ret = consumer_send_stream(sock, consumer, &msg, fds, 2);
+       ret = lttcomm_send_unix_sock(socket->fd, &msg, sizeof(msg));
        if (ret < 0) {
                goto error;
        }
 
-       health_code_update();
+       ret = consumer_recv_status_channel(socket, &key,
+                       &ua_chan->expected_stream_count);
+       if (ret < 0) {
+               goto error;
+       }
+       /* Communication protocol error. */
+       assert(key == ua_chan->key);
+       /* We need at least one where 1 stream for 1 cpu. */
+       assert(ua_chan->expected_stream_count > 0);
+
+       DBG2("UST ask channel %lu successfully done with %u stream(s)", key,
+                       ua_chan->expected_stream_count);
 
 error:
+       free(pathname);
+       health_code_update();
        return ret;
 }
 
 /*
- * Send all stream fds of UST channel to the consumer.
+ * Ask consumer to create a channel for a given session.
+ *
+ * Returns 0 on success else a negative value.
  */
-static int send_channel_streams(struct consumer_socket *sock,
-               struct ust_app_channel *uchan, struct ust_app_session *usess,
-               struct consumer_output *consumer)
+int ust_consumer_ask_channel(struct ust_app_session *ua_sess,
+               struct ust_app_channel *ua_chan, struct consumer_output *consumer,
+               struct consumer_socket *socket)
 {
        int ret;
-       char tmp_path[PATH_MAX];
-       const char *pathname;
-       struct ust_app_stream *stream, *tmp;
 
-       assert(sock);
+       assert(ua_sess);
+       assert(ua_chan);
+       assert(consumer);
+       assert(socket);
+       assert(socket->fd >= 0);
 
-       DBG("Sending streams of channel %s to UST consumer", uchan->name);
+       pthread_mutex_lock(socket->lock);
 
-       ret = send_channel(sock, uchan);
+       ret = ask_channel_creation(ua_sess, ua_chan, consumer, socket);
        if (ret < 0) {
                goto error;
        }
 
-       /* Get the right path name destination */
-       if (consumer->type == CONSUMER_DST_LOCAL) {
-               /* Set application path to the destination path */
-               ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s/%s",
-                               consumer->dst.trace_path, consumer->subdir, usess->path);
-               if (ret < 0) {
-                       PERROR("snprintf stream path");
-                       goto error;
-               }
-               pathname = tmp_path;
-               DBG3("UST local consumer tracefile path: %s", pathname);
-       } else {
-               ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
-                               consumer->subdir, usess->path);
-               if (ret < 0) {
-                       PERROR("snprintf stream path");
-                       goto error;
-               }
-               pathname = tmp_path;
-               DBG3("UST network consumer subdir path: %s", pathname);
-       }
-
-       cds_list_for_each_entry_safe(stream, tmp, &uchan->streams.head, list) {
-               if (!stream->obj->shm_fd) {
-                       continue;
-               }
-
-               ret = send_channel_stream(sock, uchan, usess, stream, consumer,
-                               pathname);
-               if (ret < 0) {
-                       goto error;
-               }
-       }
-
-       DBG("UST consumer channel streams sent");
-
-       return 0;
-
 error:
+       pthread_mutex_unlock(socket->lock);
        return ret;
 }
 
 /*
- * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
+ * Send a get channel command to consumer using the given channel key.  The
+ * channel object is populated and the stream list.
+ *
+ * Return 0 on success else a negative value.
  */
-static int send_metadata(struct consumer_socket *sock,
-               struct ust_app_session *usess, struct consumer_output *consumer)
+int ust_consumer_get_channel(struct consumer_socket *socket,
+               struct ust_app_channel *ua_chan)
 {
-       int ret, fd, fds[2];
-       char tmp_path[PATH_MAX];
-       const char *pathname;
+       int ret;
        struct lttcomm_consumer_msg msg;
 
-       /* Safety net */
-       assert(usess);
-       assert(consumer);
-       assert(sock);
-
-       if (sock->fd < 0) {
-               ERR("Consumer socket is negative (%d)", sock->fd);
-               return -EINVAL;
-       }
-
-       if (usess->metadata->obj->shm_fd == 0) {
-               ERR("Metadata obj shm_fd is 0");
-               ret = -1;
-               goto error;
-       }
+       assert(ua_chan);
+       assert(socket);
+       assert(socket->fd >= 0);
 
-       DBG("UST consumer sending metadata stream fd");
-
-       consumer_init_channel_comm_msg(&msg,
-                       LTTNG_CONSUMER_ADD_CHANNEL,
-                       usess->metadata->obj->shm_fd,
-                       usess->metadata->attr.subbuf_size,
-                       usess->metadata->obj->memory_map_size,
-                       DEFAULT_METADATA_NAME,
-                       1);
+       msg.cmd_type = LTTNG_CONSUMER_GET_CHANNEL;
+       msg.u.get_channel.key = ua_chan->key;
 
+       pthread_mutex_lock(socket->lock);
        health_code_update();
 
-       ret = consumer_send_channel(sock, &msg);
+       /* Send command and wait for OK reply. */
+       ret = consumer_send_msg(socket, &msg);
        if (ret < 0) {
                goto error;
        }
 
-       health_code_update();
-
-       /* Sending metadata shared memory fd */
-       fd = usess->metadata->obj->shm_fd;
-       ret = consumer_send_fds(sock, &fd, 1);
+       /* First, get the channel from consumer. */
+       ret = ustctl_recv_channel_from_consumer(socket->fd, &ua_chan->obj);
        if (ret < 0) {
+               if (ret != -EPIPE) {
+                       ERR("Error recv channel from consumer %d with ret %d",
+                                       socket->fd, ret);
+               } else {
+                       DBG3("UST app recv channel from consumer. Consumer is dead.");
+               }
                goto error;
        }
+       ua_chan->handle = ua_chan->obj->handle;
 
-       health_code_update();
+       /* Next, get all streams. */
+       while (1) {
+               struct ust_app_stream *stream;
 
-       /* Get correct path name destination */
-       if (consumer->type == CONSUMER_DST_LOCAL) {
-               /* Set application path to the destination path */
-               ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s/%s",
-                               consumer->dst.trace_path, consumer->subdir, usess->path);
-               if (ret < 0) {
-                       PERROR("snprintf stream path");
+               /* Create UST stream */
+               stream = ust_app_alloc_stream();
+               if (stream == NULL) {
+                       ret = -ENOMEM;
                        goto error;
                }
-               pathname = tmp_path;
 
-               /* Create directory */
-               ret = run_as_mkdir_recursive(pathname, S_IRWXU | S_IRWXG,
-                               usess->uid, usess->gid);
+               /* Stream object is populated by this call if successful. */
+               ret = ustctl_recv_stream_from_consumer(socket->fd, &stream->obj);
                if (ret < 0) {
-                       if (ret != -EEXIST) {
-                               ERR("Trace directory creation error");
-                               goto error;
+                       free(stream);
+                       if (ret == -LTTNG_UST_ERR_NOENT) {
+                               DBG3("UST app consumer has no more stream available");
+                               ret = 0;
+                               break;
+                       }
+                       if (ret != -EPIPE) {
+                               ERR("Recv stream from consumer %d with ret %d",
+                                               socket->fd, ret);
+                       } else {
+                               DBG3("UST app recv stream from consumer. Consumer is dead.");
                        }
-               }
-       } else {
-               ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
-                               consumer->subdir, usess->path);
-               if (ret < 0) {
-                       PERROR("snprintf metadata path");
                        goto error;
                }
-               pathname = tmp_path;
-       }
 
-       consumer_init_stream_comm_msg(&msg,
-                       LTTNG_CONSUMER_ADD_STREAM,
-                       usess->metadata->obj->shm_fd,
-                       usess->metadata->stream_obj->shm_fd,
-                       LTTNG_CONSUMER_ACTIVE_STREAM,
-                       DEFAULT_UST_CHANNEL_OUTPUT,
-                       usess->metadata->stream_obj->memory_map_size,
-                       usess->uid,
-                       usess->gid,
-                       consumer->net_seq_index,
-                       1, /* Flag metadata set */
-                       DEFAULT_METADATA_NAME,
-                       pathname,
-                       usess->id);
+               /* Order is important this is why a list is used. */
+               cds_list_add_tail(&stream->list, &ua_chan->streams.head);
+               ua_chan->streams.count++;
 
-       health_code_update();
+               DBG2("UST app stream %d received succesfully", ua_chan->streams.count);
+       }
+
+       /* This MUST match or else we have a synchronization problem. */
+       assert(ua_chan->expected_stream_count == ua_chan->streams.count);
 
-       /* Send stream and file descriptor */
-       fds[0] = usess->metadata->stream_obj->shm_fd;
-       fds[1] = usess->metadata->stream_obj->wait_fd;
-       ret = consumer_send_stream(sock, consumer, &msg, fds, 2);
+       /* Wait for confirmation that we can proceed with the streams. */
+       ret = consumer_recv_status_reply(socket);
        if (ret < 0) {
                goto error;
        }
 
-       health_code_update();
-
 error:
+       health_code_update();
+       pthread_mutex_unlock(socket->lock);
        return ret;
 }
 
 /*
- * Send all stream fds of the UST session to the consumer.
+ * Send a destroy channel command to consumer using the given channel key.
+ *
+ * Note that this command MUST be used prior to a successful
+ * LTTNG_CONSUMER_GET_CHANNEL because once this command is done successfully,
+ * the streams are dispatched to the consumer threads and MUST be teardown
+ * through the hang up process.
+ *
+ * Return 0 on success else a negative value.
  */
-int ust_consumer_send_session(struct ust_app_session *usess,
-               struct consumer_output *consumer, struct consumer_socket *sock)
+int ust_consumer_destroy_channel(struct consumer_socket *socket,
+               struct ust_app_channel *ua_chan)
 {
-       int ret = 0;
-       struct lttng_ht_iter iter;
-       struct ust_app_channel *ua_chan;
-
-       assert(usess);
+       int ret;
+       struct lttcomm_consumer_msg msg;
 
-       if (consumer == NULL || sock == NULL) {
-               /* There is no consumer so just ignoring the command. */
-               DBG("UST consumer does not exist. Not sending streams");
-               return 0;
-       }
+       assert(ua_chan);
+       assert(socket);
+       assert(socket->fd >= 0);
 
-       DBG("Sending metadata stream fd to consumer on %d", sock->fd);
+       msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
+       msg.u.destroy_channel.key = ua_chan->key;
 
-       pthread_mutex_lock(sock->lock);
+       pthread_mutex_lock(socket->lock);
+       health_code_update();
 
-       /* Sending metadata information to the consumer */
-       ret = send_metadata(sock, usess, consumer);
+       ret = consumer_send_msg(socket, &msg);
        if (ret < 0) {
                goto error;
        }
 
-       /* Send each channel fd streams of session */
-       rcu_read_lock();
-       cds_lfht_for_each_entry(usess->channels->ht, &iter.iter, ua_chan,
-                       node.node) {
-               /*
-                * Indicate that the channel was not created on the tracer side so skip
-                * sending unexisting streams.
-                */
-               if (ua_chan->obj == NULL) {
-                       continue;
-               }
+error:
+       health_code_update();
+       pthread_mutex_unlock(socket->lock);
+       return ret;
+}
 
-               ret = send_channel_streams(sock, ua_chan, usess, consumer);
-               if (ret < 0) {
-                       rcu_read_unlock();
-                       goto error;
+/*
+ * Send a given stream to UST tracer.
+ *
+ * On success return 0 else a negative value.
+ */
+int ust_consumer_send_stream_to_ust(struct ust_app *app,
+               struct ust_app_channel *channel, struct ust_app_stream *stream)
+{
+       int ret;
+
+       assert(app);
+       assert(stream);
+       assert(channel);
+
+       DBG2("UST consumer send stream to app %d", app->sock);
+
+       /* Relay stream to application. */
+       ret = ustctl_send_stream_to_ust(app->sock, channel->obj, stream->obj);
+       if (ret < 0) {
+               if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("Error ustctl send stream %s to app pid: %d with ret %d",
+                                       stream->name, app->pid, ret);
+               } else {
+                       DBG3("UST app send stream to ust failed. Application is dead.");
                }
+               goto error;
        }
-       rcu_read_unlock();
 
-       DBG("consumer fds (metadata and channel streams) sent");
+error:
+       return ret;
+}
+
+/*
+ * Send channel previously received from the consumer to the UST tracer.
+ *
+ * On success return 0 else a negative value.
+ */
+int ust_consumer_send_channel_to_ust(struct ust_app *app,
+               struct ust_app_session *ua_sess, struct ust_app_channel *channel)
+{
+       int ret;
+
+       assert(app);
+       assert(ua_sess);
+       assert(channel);
+       assert(channel->obj);
+
+       DBG2("UST app send channel to app sock %d pid %d (name: %s, key: %lu)",
+                       app->sock, app->pid, channel->name, channel->key);
 
-       /* All good! */
-       ret = 0;
+       /* Send stream to application. */
+       ret = ustctl_send_channel_to_ust(app->sock, ua_sess->handle, channel->obj);
+       if (ret < 0) {
+               if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
+                       ERR("Error ustctl send channel %s to app pid: %d with ret %d",
+                                       channel->name, app->pid, ret);
+               } else {
+                       DBG3("UST app send channel to ust failed. Application is dead.");
+               }
+               goto error;
+       }
 
 error:
-       pthread_mutex_unlock(sock->lock);
        return ret;
 }
index b4a711b0ae1dbe4c731d8d81e82b514615ecef39..f48ea42f27bfdfaff0e34cee692f3364ad590817 100644 (file)
 #include "consumer.h"
 #include "ust-app.h"
 
-int ust_consumer_send_session(struct ust_app_session *usess,
-               struct consumer_output *consumer, struct consumer_socket *sock);
+int ust_consumer_ask_channel(struct ust_app_session *ua_sess,
+               struct ust_app_channel *ua_chan, struct consumer_output *consumer,
+               struct consumer_socket *socket);
+
+int ust_consumer_get_channel(struct consumer_socket *socket,
+               struct ust_app_channel *ua_chan);
+
+int ust_consumer_destroy_channel(struct consumer_socket *socket,
+               struct ust_app_channel *ua_chan);
+
+int ust_consumer_send_stream_to_ust(struct ust_app *app,
+               struct ust_app_channel *channel, struct ust_app_stream *stream);
+
+int ust_consumer_send_channel_to_ust(struct ust_app *app,
+               struct ust_app_session *ua_sess, struct ust_app_channel *channel);
 
 #endif /* _UST_CONSUMER_H */
index f91259c2a8729b334747b58528308a49293734ee..c3a947aaaf61094bdf98298dfabd8d92db81222a 100644 (file)
@@ -13,6 +13,7 @@ noinst_LTLIBRARIES = libcommon.la
 
 libcommon_la_SOURCES = error.h error.c utils.c utils.h runas.c runas.h \
                        common.h futex.c futex.h uri.c uri.h defaults.c
+libcommon_la_LIBADD = -luuid
 
 # Consumer library
 noinst_LTLIBRARIES += libconsumer.la
index 2d832822f538432933d32a61353acc3fb01cb7de..537375b91f5ed6244f0013b499e0e3586b52de32 100644 (file)
@@ -9,4 +9,4 @@ COMPAT=compat-poll.c
 endif
 
 libcompat_la_SOURCES = poll.h fcntl.h endian.h mman.h clone.h \
-                       socket.h compat-fcntl.c $(COMPAT)
+                       socket.h compat-fcntl.c uuid.h tid.h $(COMPAT)
diff --git a/src/common/compat/tid.h b/src/common/compat/tid.h
new file mode 100644 (file)
index 0000000..40f562f
--- /dev/null
@@ -0,0 +1,51 @@
+/*
+ * Copyright 2012 (c) - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ *
+ * gettid compatibility layer.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to
+ * deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
+ * sell copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+ * IN THE SOFTWARE.
+ */
+
+#ifndef LTTNG_TID_H
+#define LTTNG_TID_H
+
+#ifdef __linux__
+#include <syscall.h>
+#endif
+
+#if defined(_syscall0)
+_syscall0(pid_t, gettid)
+#elif defined(__NR_gettid)
+#include <unistd.h>
+static inline pid_t gettid(void)
+{
+       return syscall(__NR_gettid);
+}
+#else
+#include <sys/types.h>
+#include <unistd.h>
+
+/* Fall-back on getpid for tid if not available. */
+static inline pid_t gettid(void)
+{
+       return getpid();
+}
+#endif
+
+#endif /* LTTNG_TID_H */
diff --git a/src/common/compat/uuid.h b/src/common/compat/uuid.h
new file mode 100644 (file)
index 0000000..5bf8beb
--- /dev/null
@@ -0,0 +1,62 @@
+/*
+ * Copyright (C) 2011  Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to
+ * deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
+ * sell copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+ * IN THE SOFTWARE.
+ */
+
+#ifndef LTTNG_UUID_H
+#define LTTNG_UUID_H
+
+#include <config.h>
+/*
+ * Includes final \0.
+ */
+#define UUID_STR_LEN           37
+
+#ifdef LTTNG_HAVE_LIBUUID
+#include <uuid/uuid.h>
+
+static inline
+int lttng_uuid_generate(unsigned char *uuid_out)
+{
+       uuid_generate(uuid_out);
+       return 0;
+}
+
+#elif defined(LTTNG_HAVE_LIBC_UUID)
+#include <uuid.h>
+#include <stdint.h>
+
+static inline
+int lttng_uuid_generate(unsigned char *uuid_out)
+{
+       uint32_t status;
+
+       uuid_create((uuid_t *) uuid_out, &status);
+       if (status == uuid_s_ok)
+               return 0;
+       else
+               return -1;
+}
+
+#else
+#error "LTTng-Tools needs to have a UUID generator configured."
+#endif
+
+#endif /* LTTNG_UUID_H */
index 08592f6c0b8713f1ce1765a7de4d9237fd8bd81a..09b3bee330457d46483a85d1f2f5402e25241b9e 100644 (file)
@@ -82,7 +82,7 @@ static void notify_thread_pipe(int wpipe)
  * Find a stream. The consumer_data.lock must be locked during this
  * call.
  */
-static struct lttng_consumer_stream *consumer_find_stream(int key,
+static struct lttng_consumer_stream *find_stream(int key,
                struct lttng_ht *ht)
 {
        struct lttng_ht_iter iter;
@@ -109,12 +109,12 @@ static struct lttng_consumer_stream *consumer_find_stream(int key,
        return stream;
 }
 
-void consumer_steal_stream_key(int key, struct lttng_ht *ht)
+static void steal_stream_key(int key, struct lttng_ht *ht)
 {
        struct lttng_consumer_stream *stream;
 
        rcu_read_lock();
-       stream = consumer_find_stream(key, ht);
+       stream = find_stream(key, ht);
        if (stream) {
                stream->key = -1;
                /*
@@ -133,7 +133,7 @@ void consumer_steal_stream_key(int key, struct lttng_ht *ht)
  * RCU read side lock MUST be acquired before calling this function and
  * protects the channel ptr.
  */
-static struct lttng_consumer_channel *consumer_find_channel(int key)
+struct lttng_consumer_channel *consumer_find_channel(unsigned long key)
 {
        struct lttng_ht_iter iter;
        struct lttng_ht_node_ulong *node;
@@ -144,8 +144,7 @@ static struct lttng_consumer_channel *consumer_find_channel(int key)
                return NULL;
        }
 
-       lttng_ht_lookup(consumer_data.channel_ht, (void *)((unsigned long) key),
-                       &iter);
+       lttng_ht_lookup(consumer_data.channel_ht, (void *) key, &iter);
        node = lttng_ht_iter_get_node_ulong(&iter);
        if (node != NULL) {
                channel = caa_container_of(node, struct lttng_consumer_channel, node);
@@ -154,39 +153,30 @@ static struct lttng_consumer_channel *consumer_find_channel(int key)
        return channel;
 }
 
-static void consumer_steal_channel_key(int key)
+static void free_stream_rcu(struct rcu_head *head)
 {
-       struct lttng_consumer_channel *channel;
+       struct lttng_ht_node_ulong *node =
+               caa_container_of(head, struct lttng_ht_node_ulong, head);
+       struct lttng_consumer_stream *stream =
+               caa_container_of(node, struct lttng_consumer_stream, node);
 
-       rcu_read_lock();
-       channel = consumer_find_channel(key);
-       if (channel) {
-               channel->key = -1;
-               /*
-                * We don't want the lookup to match, but we still need
-                * to iterate on this channel when iterating over the hash table. Just
-                * change the node key.
-                */
-               channel->node.key = -1;
-       }
-       rcu_read_unlock();
+       free(stream);
 }
 
-static
-void consumer_free_stream(struct rcu_head *head)
+static void free_channel_rcu(struct rcu_head *head)
 {
        struct lttng_ht_node_ulong *node =
                caa_container_of(head, struct lttng_ht_node_ulong, head);
-       struct lttng_consumer_stream *stream =
-               caa_container_of(node, struct lttng_consumer_stream, node);
+       struct lttng_consumer_channel *channel =
+               caa_container_of(node, struct lttng_consumer_channel, node);
 
-       free(stream);
+       free(channel);
 }
 
 /*
  * RCU protected relayd socket pair free.
  */
-static void consumer_rcu_free_relayd(struct rcu_head *head)
+static void free_relayd_rcu(struct rcu_head *head)
 {
        struct lttng_ht_node_ulong *node =
                caa_container_of(head, struct lttng_ht_node_ulong, head);
@@ -231,7 +221,44 @@ static void destroy_relayd(struct consumer_relayd_sock_pair *relayd)
        }
 
        /* RCU free() call */
-       call_rcu(&relayd->node.head, consumer_rcu_free_relayd);
+       call_rcu(&relayd->node.head, free_relayd_rcu);
+}
+
+/*
+ * Remove a channel from the global list protected by a mutex. This function is
+ * also responsible for freeing its data structures.
+ */
+void consumer_del_channel(struct lttng_consumer_channel *channel)
+{
+       int ret;
+       struct lttng_ht_iter iter;
+
+       DBG("Consumer delete channel key %d", channel->key);
+
+       pthread_mutex_lock(&consumer_data.lock);
+
+       switch (consumer_data.type) {
+       case LTTNG_CONSUMER_KERNEL:
+               break;
+       case LTTNG_CONSUMER32_UST:
+       case LTTNG_CONSUMER64_UST:
+               lttng_ustconsumer_del_channel(channel);
+               break;
+       default:
+               ERR("Unknown consumer_data type");
+               assert(0);
+               goto end;
+       }
+
+       rcu_read_lock();
+       iter.iter.node = &channel->node.node;
+       ret = lttng_ht_del(consumer_data.channel_ht, &iter);
+       assert(!ret);
+       rcu_read_unlock();
+
+       call_rcu(&channel->node.head, free_channel_rcu);
+end:
+       pthread_mutex_unlock(&consumer_data.lock);
 }
 
 /*
@@ -368,7 +395,7 @@ void consumer_del_stream(struct lttng_consumer_stream *stream,
 
        if (ht == NULL) {
                /* Means the stream was allocated but not successfully added */
-               goto free_stream;
+               goto free_stream_rcu;
        }
 
        pthread_mutex_lock(&consumer_data.lock);
@@ -413,18 +440,6 @@ void consumer_del_stream(struct lttng_consumer_stream *stream,
                        PERROR("close");
                }
        }
-       if (stream->wait_fd >= 0 && !stream->wait_fd_is_copy) {
-               ret = close(stream->wait_fd);
-               if (ret) {
-                       PERROR("close");
-               }
-       }
-       if (stream->shm_fd >= 0 && stream->wait_fd != stream->shm_fd) {
-               ret = close(stream->shm_fd);
-               if (ret) {
-                       PERROR("close");
-               }
-       }
 
        /* Check and cleanup relayd */
        rcu_read_lock();
@@ -458,7 +473,7 @@ void consumer_del_stream(struct lttng_consumer_stream *stream,
 
        uatomic_dec(&stream->chan->refcount);
        if (!uatomic_read(&stream->chan->refcount)
-                       && !uatomic_read(&stream->chan->nb_init_streams)) {
+                       && !uatomic_read(&stream->chan->nb_init_stream_left)) {
                free_chan = stream->chan;
        }
 
@@ -471,91 +486,67 @@ end:
                consumer_del_channel(free_chan);
        }
 
-free_stream:
-       call_rcu(&stream->node.head, consumer_free_stream);
+free_stream_rcu:
+       call_rcu(&stream->node.head, free_stream_rcu);
 }
 
-struct lttng_consumer_stream *consumer_allocate_stream(
-               int channel_key, int stream_key,
-               int shm_fd, int wait_fd,
+struct lttng_consumer_stream *consumer_allocate_stream(int channel_key,
+               int stream_key,
                enum lttng_consumer_stream_state state,
-               uint64_t mmap_len,
-               enum lttng_event_output output,
-               const char *path_name,
+               const char *channel_name,
                uid_t uid,
                gid_t gid,
-               int net_index,
-               int metadata_flag,
+               int relayd_id,
                uint64_t session_id,
-               int *alloc_ret)
+               int cpu,
+               int *alloc_ret,
+               enum consumer_channel_type type)
 {
+       int ret;
        struct lttng_consumer_stream *stream;
 
        stream = zmalloc(sizeof(*stream));
        if (stream == NULL) {
                PERROR("malloc struct lttng_consumer_stream");
-               *alloc_ret = -ENOMEM;
+               ret = -ENOMEM;
                goto end;
        }
 
        rcu_read_lock();
 
-       /*
-        * Get stream's channel reference. Needed when adding the stream to the
-        * global hash table.
-        */
-       stream->chan = consumer_find_channel(channel_key);
-       if (!stream->chan) {
-               *alloc_ret = -ENOENT;
-               ERR("Unable to find channel for stream %d", stream_key);
-               goto error;
-       }
-
        stream->key = stream_key;
-       stream->shm_fd = shm_fd;
-       stream->wait_fd = wait_fd;
        stream->out_fd = -1;
        stream->out_fd_offset = 0;
        stream->state = state;
-       stream->mmap_len = mmap_len;
-       stream->mmap_base = NULL;
-       stream->output = output;
        stream->uid = uid;
        stream->gid = gid;
-       stream->net_seq_idx = net_index;
-       stream->metadata_flag = metadata_flag;
+       stream->net_seq_idx = relayd_id;
        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
-        * internal hash table to match streams in the metadata_ht to the epoll set
-        * file descriptor.
-        */
-       if (metadata_flag) {
-               lttng_ht_node_init_ulong(&stream->node, stream->wait_fd);
+       /* If channel is the metadata, flag this stream as metadata. */
+       if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
+               stream->metadata_flag = 1;
+               /* Metadata is flat out. */
+               strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
        } else {
-               lttng_ht_node_init_ulong(&stream->node, stream->key);
+               /* Format stream name to <channel_name>_<cpu_number> */
+               ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
+                               channel_name, cpu);
+               if (ret < 0) {
+                       PERROR("snprintf stream name");
+                       goto error;
+               }
        }
 
+       /* Key is always the wait_fd for streams. */
+       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.
-        */
-       pthread_mutex_lock(&consumer_data.lock);
-       stream->cpu = stream->chan->cpucount++;
-       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, 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->session_id);
+       DBG3("Allocated stream %s (key %d, relayd_id %d, session_id %" PRIu64,
+                       stream->name, stream->key, stream->net_seq_idx, stream->session_id);
 
        rcu_read_unlock();
        return stream;
@@ -564,13 +555,16 @@ error:
        rcu_read_unlock();
        free(stream);
 end:
+       if (alloc_ret) {
+               *alloc_ret = ret;
+       }
        return NULL;
 }
 
 /*
  * Add a stream to the global list protected by a mutex.
  */
-static int consumer_add_stream(struct lttng_consumer_stream *stream,
+static int add_stream(struct lttng_consumer_stream *stream,
                struct lttng_ht *ht)
 {
        int ret = 0;
@@ -586,7 +580,7 @@ static int consumer_add_stream(struct lttng_consumer_stream *stream,
        rcu_read_lock();
 
        /* Steal stream identifier to avoid having streams with the same key */
-       consumer_steal_stream_key(stream->key, ht);
+       steal_stream_key(stream->key, ht);
 
        lttng_ht_add_unique_ulong(ht, &stream->node);
 
@@ -607,14 +601,14 @@ static int consumer_add_stream(struct lttng_consumer_stream *stream,
        uatomic_inc(&stream->chan->refcount);
 
        /*
-        * When nb_init_streams reaches 0, we don't need to trigger any action in
-        * terms of destroying the associated channel, because the action that
+        * When nb_init_stream_left reaches 0, we don't need to trigger any action
+        * in terms of destroying the associated channel, because the action that
         * causes the count to become 0 also causes a stream to be added. The
         * channel deletion will thus be triggered by the following removal of this
         * stream.
         */
-       if (uatomic_read(&stream->chan->nb_init_streams) > 0) {
-               uatomic_dec(&stream->chan->nb_init_streams);
+       if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
+               uatomic_dec(&stream->chan->nb_init_stream_left);
        }
 
        /* Update consumer data once the node is inserted. */
@@ -638,16 +632,12 @@ static int add_relayd(struct consumer_relayd_sock_pair *relayd)
        struct lttng_ht_node_ulong *node;
        struct lttng_ht_iter iter;
 
-       if (relayd == NULL) {
-               ret = -1;
-               goto end;
-       }
+       assert(relayd);
 
        lttng_ht_lookup(consumer_data.relayd_ht,
                        (void *)((unsigned long) relayd->net_seq_idx), &iter);
        node = lttng_ht_iter_get_node_ulong(&iter);
        if (node != NULL) {
-               /* Relayd already exist. Ignore the insertion */
                goto end;
        }
        lttng_ht_add_unique_ulong(consumer_data.relayd_ht, &relayd->node);
@@ -774,119 +764,48 @@ error:
        return outfd;
 }
 
-static
-void consumer_free_channel(struct rcu_head *head)
-{
-       struct lttng_ht_node_ulong *node =
-               caa_container_of(head, struct lttng_ht_node_ulong, head);
-       struct lttng_consumer_channel *channel =
-               caa_container_of(node, struct lttng_consumer_channel, node);
-
-       free(channel);
-}
-
 /*
- * Remove a channel from the global list protected by a mutex. This
- * function is also responsible for freeing its data structures.
+ * Allocate and return a new lttng_consumer_channel object using the given key
+ * to initialize the hash table node.
+ *
+ * On error, return NULL.
  */
-void consumer_del_channel(struct lttng_consumer_channel *channel)
-{
-       int ret;
-       struct lttng_ht_iter iter;
-
-       DBG("Consumer delete channel key %d", channel->key);
-
-       pthread_mutex_lock(&consumer_data.lock);
-
-       switch (consumer_data.type) {
-       case LTTNG_CONSUMER_KERNEL:
-               break;
-       case LTTNG_CONSUMER32_UST:
-       case LTTNG_CONSUMER64_UST:
-               lttng_ustconsumer_del_channel(channel);
-               break;
-       default:
-               ERR("Unknown consumer_data type");
-               assert(0);
-               goto end;
-       }
-
-       rcu_read_lock();
-       iter.iter.node = &channel->node.node;
-       ret = lttng_ht_del(consumer_data.channel_ht, &iter);
-       assert(!ret);
-       rcu_read_unlock();
-
-       if (channel->mmap_base != NULL) {
-               ret = munmap(channel->mmap_base, channel->mmap_len);
-               if (ret != 0) {
-                       PERROR("munmap");
-               }
-       }
-       if (channel->wait_fd >= 0 && !channel->wait_fd_is_copy) {
-               ret = close(channel->wait_fd);
-               if (ret) {
-                       PERROR("close");
-               }
-       }
-       if (channel->shm_fd >= 0 && channel->wait_fd != channel->shm_fd) {
-               ret = close(channel->shm_fd);
-               if (ret) {
-                       PERROR("close");
-               }
-       }
-
-       call_rcu(&channel->node.head, consumer_free_channel);
-end:
-       pthread_mutex_unlock(&consumer_data.lock);
-}
-
-struct lttng_consumer_channel *consumer_allocate_channel(
-               int channel_key,
-               int shm_fd, int wait_fd,
-               uint64_t mmap_len,
-               uint64_t max_sb_size,
-               unsigned int nb_init_streams)
+struct lttng_consumer_channel *consumer_allocate_channel(unsigned long key,
+               uint64_t session_id,
+               const char *pathname,
+               const char *name,
+               uid_t uid,
+               gid_t gid,
+               int relayd_id,
+               enum lttng_event_output output)
 {
        struct lttng_consumer_channel *channel;
-       int ret;
 
        channel = zmalloc(sizeof(*channel));
        if (channel == NULL) {
                PERROR("malloc struct lttng_consumer_channel");
                goto end;
        }
-       channel->key = channel_key;
-       channel->shm_fd = shm_fd;
-       channel->wait_fd = wait_fd;
-       channel->mmap_len = mmap_len;
-       channel->max_sb_size = max_sb_size;
+
+       channel->key = key;
        channel->refcount = 0;
-       channel->nb_init_streams = nb_init_streams;
+       channel->session_id = session_id;
+       channel->uid = uid;
+       channel->gid = gid;
+       channel->relayd_id = relayd_id;
+       channel->output = output;
+
+       strncpy(channel->pathname, pathname, sizeof(channel->pathname));
+       channel->pathname[sizeof(channel->pathname) - 1] = '\0';
+
+       strncpy(channel->name, name, sizeof(channel->name));
+       channel->name[sizeof(channel->name) - 1] = '\0';
+
        lttng_ht_node_init_ulong(&channel->node, channel->key);
+       CDS_INIT_LIST_HEAD(&channel->streams.head);
+
+       DBG("Allocated channel (key %d)", channel->key)
 
-       switch (consumer_data.type) {
-       case LTTNG_CONSUMER_KERNEL:
-               channel->mmap_base = NULL;
-               channel->mmap_len = 0;
-               break;
-       case LTTNG_CONSUMER32_UST:
-       case LTTNG_CONSUMER64_UST:
-               ret = lttng_ustconsumer_allocate_channel(channel);
-               if (ret) {
-                       free(channel);
-                       return NULL;
-               }
-               break;
-       default:
-               ERR("Unknown consumer_data type");
-               assert(0);
-               goto end;
-       }
-       DBG("Allocated channel (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, max_sb_size %llu)",
-                       channel->key, channel->shm_fd, channel->wait_fd,
-                       (unsigned long long) channel->mmap_len,
-                       (unsigned long long) channel->max_sb_size);
 end:
        return channel;
 }
@@ -896,12 +815,11 @@ end:
  */
 int consumer_add_channel(struct lttng_consumer_channel *channel)
 {
+       int ret = 0;
        struct lttng_ht_node_ulong *node;
        struct lttng_ht_iter iter;
 
        pthread_mutex_lock(&consumer_data.lock);
-       /* Steal channel identifier, for UST */
-       consumer_steal_channel_key(channel->key);
        rcu_read_lock();
 
        lttng_ht_lookup(consumer_data.channel_ht,
@@ -909,6 +827,8 @@ int consumer_add_channel(struct lttng_consumer_channel *channel)
        node = lttng_ht_iter_get_node_ulong(&iter);
        if (node != NULL) {
                /* Channel already exist. Ignore the insertion */
+               ERR("Consumer add channel key %d already exists!", channel->key);
+               ret = -1;
                goto end;
        }
 
@@ -918,7 +838,7 @@ end:
        rcu_read_unlock();
        pthread_mutex_unlock(&consumer_data.lock);
 
-       return 0;
+       return ret;
 }
 
 /*
@@ -928,14 +848,19 @@ end:
  *
  * Returns the number of fds in the structures.
  */
-static int consumer_update_poll_array(
-               struct lttng_consumer_local_data *ctx, struct pollfd **pollfd,
-               struct lttng_consumer_stream **local_stream, struct lttng_ht *ht)
+static int update_poll_array(struct lttng_consumer_local_data *ctx,
+               struct pollfd **pollfd, struct lttng_consumer_stream **local_stream,
+               struct lttng_ht *ht)
 {
        int i = 0;
        struct lttng_ht_iter iter;
        struct lttng_consumer_stream *stream;
 
+       assert(ctx);
+       assert(ht);
+       assert(pollfd);
+       assert(local_stream);
+
        DBG("Updating poll fd array");
        rcu_read_lock();
        cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
@@ -1002,8 +927,8 @@ exit:
 /*
  * Set the error socket.
  */
-void lttng_consumer_set_error_sock(
-               struct lttng_consumer_local_data *ctx, int sock)
+void lttng_consumer_set_error_sock(struct lttng_consumer_local_data *ctx,
+               int sock)
 {
        ctx->consumer_error_socket = sock;
 }
@@ -1021,8 +946,7 @@ void lttng_consumer_set_command_sock_path(
  * Send return code to the session daemon.
  * If the socket is not defined, we return 0, it is not a fatal error
  */
-int lttng_consumer_send_error(
-               struct lttng_consumer_local_data *ctx, int cmd)
+int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, int cmd)
 {
        if (ctx->consumer_error_socket > 0) {
                return lttcomm_send_unix_sock(ctx->consumer_error_socket, &cmd,
@@ -1039,14 +963,12 @@ int lttng_consumer_send_error(
 void lttng_consumer_cleanup(void)
 {
        struct lttng_ht_iter iter;
-       struct lttng_ht_node_ulong *node;
+       struct lttng_consumer_channel *channel;
 
        rcu_read_lock();
 
-       cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, node,
-                       node) {
-               struct lttng_consumer_channel *channel =
-                       caa_container_of(node, struct lttng_consumer_channel, node);
+       cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, channel,
+                       node.node) {
                consumer_del_channel(channel);
        }
 
@@ -1092,11 +1014,11 @@ void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
         * Don't care about error values, as these are just hints and ways to
         * limit the amount of page cache used.
         */
-       if (orig_offset < stream->chan->max_sb_size) {
+       if (orig_offset < stream->max_sb_size) {
                return;
        }
-       lttng_sync_file_range(outfd, orig_offset - stream->chan->max_sb_size,
-                       stream->chan->max_sb_size,
+       lttng_sync_file_range(outfd, orig_offset - stream->max_sb_size,
+                       stream->max_sb_size,
                        SYNC_FILE_RANGE_WAIT_BEFORE
                        | SYNC_FILE_RANGE_WRITE
                        | SYNC_FILE_RANGE_WAIT_AFTER);
@@ -1114,8 +1036,8 @@ void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
         * defined. So it can be expected to lead to lower throughput in
         * streaming.
         */
-       posix_fadvise(outfd, orig_offset - stream->chan->max_sb_size,
-                       stream->chan->max_sb_size, POSIX_FADV_DONTNEED);
+       posix_fadvise(outfd, orig_offset - stream->max_sb_size,
+                       stream->max_sb_size, POSIX_FADV_DONTNEED);
 }
 
 /*
@@ -1281,8 +1203,7 @@ void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
  */
 static int write_relayd_metadata_id(int fd,
                struct lttng_consumer_stream *stream,
-               struct consumer_relayd_sock_pair *relayd,
-               unsigned long padding)
+               struct consumer_relayd_sock_pair *relayd, unsigned long padding)
 {
        int ret;
        struct lttcomm_relayd_metadata_payload hdr;
@@ -1334,6 +1255,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
                unsigned long padding)
 {
        unsigned long mmap_offset;
+       void *mmap_base;
        ssize_t ret = 0, written = 0;
        off_t orig_offset = stream->out_fd_offset;
        /* Default is on the disk */
@@ -1355,12 +1277,18 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
        /* get the offset inside the fd to mmap */
        switch (consumer_data.type) {
        case LTTNG_CONSUMER_KERNEL:
+               mmap_base = stream->mmap_base;
                ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset);
                break;
        case LTTNG_CONSUMER32_UST:
        case LTTNG_CONSUMER64_UST:
-               ret = lttng_ustctl_get_mmap_read_offset(stream->chan->handle,
-                               stream->buf, &mmap_offset);
+               mmap_base = lttng_ustctl_get_mmap_base(stream);
+               if (!mmap_base) {
+                       ERR("read mmap get mmap base for stream %s", stream->name);
+                       written = -1;
+                       goto end;
+               }
+               ret = lttng_ustctl_get_mmap_read_offset(stream, &mmap_offset);
                break;
        default:
                ERR("Unknown consumer_data type");
@@ -1420,7 +1348,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 
        while (len > 0) {
                do {
-                       ret = write(outfd, stream->mmap_base + mmap_offset, len);
+                       ret = write(outfd, mmap_base + mmap_offset, len);
                } while (ret < 0 && errno == EINTR);
                DBG("Consumer mmap write() ret %zd (len %lu)", ret, len);
                if (ret < 0) {
@@ -1693,15 +1621,14 @@ end:
  *
  * Returns 0 on success, < 0 on error
  */
-int lttng_consumer_take_snapshot(struct lttng_consumer_local_data *ctx,
-               struct lttng_consumer_stream *stream)
+int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream)
 {
        switch (consumer_data.type) {
        case LTTNG_CONSUMER_KERNEL:
-               return lttng_kconsumer_take_snapshot(ctx, stream);
+               return lttng_kconsumer_take_snapshot(stream);
        case LTTNG_CONSUMER32_UST:
        case LTTNG_CONSUMER64_UST:
-               return lttng_ustconsumer_take_snapshot(ctx, stream);
+               return lttng_ustconsumer_take_snapshot(stream);
        default:
                ERR("Unknown consumer_data type");
                assert(0);
@@ -1715,17 +1642,15 @@ int lttng_consumer_take_snapshot(struct lttng_consumer_local_data *ctx,
  *
  * Returns 0 on success, < 0 on error
  */
-int lttng_consumer_get_produced_snapshot(
-               struct lttng_consumer_local_data *ctx,
-               struct lttng_consumer_stream *stream,
+int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
                unsigned long *pos)
 {
        switch (consumer_data.type) {
        case LTTNG_CONSUMER_KERNEL:
-               return lttng_kconsumer_get_produced_snapshot(ctx, stream, pos);
+               return lttng_kconsumer_get_produced_snapshot(stream, pos);
        case LTTNG_CONSUMER32_UST:
        case LTTNG_CONSUMER64_UST:
-               return lttng_ustconsumer_get_produced_snapshot(ctx, stream, pos);
+               return lttng_ustconsumer_get_produced_snapshot(stream, pos);
        default:
                ERR("Unknown consumer_data type");
                assert(0);
@@ -1825,7 +1750,7 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
 
        if (ht == NULL) {
                /* Means the stream was allocated but not successfully added */
-               goto free_stream;
+               goto free_stream_rcu;
        }
 
        pthread_mutex_lock(&consumer_data.lock);
@@ -1868,20 +1793,6 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
                }
        }
 
-       if (stream->wait_fd >= 0 && !stream->wait_fd_is_copy) {
-               ret = close(stream->wait_fd);
-               if (ret) {
-                       PERROR("close");
-               }
-       }
-
-       if (stream->shm_fd >= 0 && stream->wait_fd != stream->shm_fd) {
-               ret = close(stream->shm_fd);
-               if (ret) {
-                       PERROR("close");
-               }
-       }
-
        /* Check and cleanup relayd */
        rcu_read_lock();
        relayd = consumer_find_relayd(stream->net_seq_idx);
@@ -1914,7 +1825,7 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
        /* Atomically decrement channel refcount since other threads can use it. */
        uatomic_dec(&stream->chan->refcount);
        if (!uatomic_read(&stream->chan->refcount)
-                       && !uatomic_read(&stream->chan->nb_init_streams)) {
+                       && !uatomic_read(&stream->chan->nb_init_stream_left)) {
                /* Go for channel deletion! */
                free_chan = stream->chan;
        }
@@ -1927,15 +1838,15 @@ end:
                consumer_del_channel(free_chan);
        }
 
-free_stream:
-       call_rcu(&stream->node.head, consumer_free_stream);
+free_stream_rcu:
+       call_rcu(&stream->node.head, free_stream_rcu);
 }
 
 /*
  * Action done with the metadata stream when adding it to the consumer internal
  * data structures to handle it.
  */
-static int consumer_add_metadata_stream(struct lttng_consumer_stream *stream,
+static int add_metadata_stream(struct lttng_consumer_stream *stream,
                struct lttng_ht *ht)
 {
        int ret = 0;
@@ -1946,7 +1857,7 @@ static int consumer_add_metadata_stream(struct lttng_consumer_stream *stream,
        assert(stream);
        assert(ht);
 
-       DBG3("Adding metadata stream %d to hash table", stream->wait_fd);
+       DBG3("Adding metadata stream %d to hash table", stream->key);
 
        pthread_mutex_lock(&consumer_data.lock);
        pthread_mutex_lock(&stream->lock);
@@ -1962,7 +1873,7 @@ static int consumer_add_metadata_stream(struct lttng_consumer_stream *stream,
         * Lookup the stream just to make sure it does not exist in our internal
         * state. This should NEVER happen.
         */
-       lttng_ht_lookup(ht, (void *)((unsigned long) stream->wait_fd), &iter);
+       lttng_ht_lookup(ht, (void *)((unsigned long) stream->key), &iter);
        node = lttng_ht_iter_get_node_ulong(&iter);
        assert(!node);
 
@@ -1976,14 +1887,14 @@ static int consumer_add_metadata_stream(struct lttng_consumer_stream *stream,
        uatomic_inc(&stream->chan->refcount);
 
        /*
-        * When nb_init_streams reaches 0, we don't need to trigger any action in
-        * terms of destroying the associated channel, because the action that
+        * When nb_init_stream_left reaches 0, we don't need to trigger any action
+        * in terms of destroying the associated channel, because the action that
         * causes the count to become 0 also causes a stream to be added. The
         * channel deletion will thus be triggered by the following removal of this
         * stream.
         */
-       if (uatomic_read(&stream->chan->nb_init_streams) > 0) {
-               uatomic_dec(&stream->chan->nb_init_streams);
+       if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
+               uatomic_dec(&stream->chan->nb_init_stream_left);
        }
 
        lttng_ht_add_unique_ulong(ht, &stream->node);
@@ -2163,7 +2074,7 @@ restart:
                                        DBG("Adding metadata stream %d to poll set",
                                                        stream->wait_fd);
 
-                                       ret = consumer_add_metadata_stream(stream, metadata_ht);
+                                       ret = add_metadata_stream(stream, metadata_ht);
                                        if (ret) {
                                                ERR("Unable to add metadata stream");
                                                /* Stream was not setup properly. Continuing. */
@@ -2305,7 +2216,7 @@ void *consumer_thread_data_poll(void *data)
                                pthread_mutex_unlock(&consumer_data.lock);
                                goto end;
                        }
-                       ret = consumer_update_poll_array(ctx, &pollfd, local_stream,
+                       ret = update_poll_array(ctx, &pollfd, local_stream,
                                        data_ht);
                        if (ret < 0) {
                                ERR("Error in allocating pollfd or local_outfds");
@@ -2372,7 +2283,7 @@ void *consumer_thread_data_poll(void *data)
                                continue;
                        }
 
-                       ret = consumer_add_stream(new_stream, data_ht);
+                       ret = add_stream(new_stream, data_ht);
                        if (ret) {
                                ERR("Consumer add stream %d failed. Continuing",
                                                new_stream->key);
@@ -2605,10 +2516,10 @@ void *consumer_thread_sessiond_poll(void *data)
                        DBG("consumer_thread_receive_fds received quit from signal");
                        goto end;
                }
-               DBG("received fds on sock");
+               DBG("received command on sock");
        }
 end:
-       DBG("consumer_thread_receive_fds exiting");
+       DBG("Consumer thread sessiond poll exiting");
 
        /*
         * when all fds have hung up, the polling thread
@@ -2741,7 +2652,7 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
                lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
                ret = -1;
                fd = -1;        /* Just in case it gets set with an invalid value. */
-               goto error;
+               goto error_close;
        }
 
        /* We have the fds without error. Send status back. */
@@ -2784,6 +2695,13 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
                        pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
                }
                if (ret < 0) {
+                       /*
+                        * Close all sockets of a relayd object. It will be freed if it was
+                        * created at the error code path or else it will be garbage
+                        * collect.
+                        */
+                       (void) relayd_close(&relayd->control_sock);
+                       (void) relayd_close(&relayd->data_sock);
                        goto error;
                }
 
@@ -2833,10 +2751,8 @@ error:
                }
        }
 
+error_close:
        if (relayd_created) {
-               /* We just want to cleanup. Ignore ret value. */
-               (void) relayd_close(&relayd->control_sock);
-               (void) relayd_close(&relayd->data_sock);
                free(relayd);
        }
 
@@ -3044,3 +2960,26 @@ int consumer_send_status_msg(int sock, int ret_code)
 
        return lttcomm_send_unix_sock(sock, &msg, sizeof(msg));
 }
+
+/*
+ * Send a channel status message to the sessiond daemon.
+ *
+ * Return the sendmsg() return value.
+ */
+int consumer_send_status_channel(int sock,
+               struct lttng_consumer_channel *channel)
+{
+       struct lttcomm_consumer_status_channel msg;
+
+       assert(sock >= 0);
+
+       if (!channel) {
+               msg.ret_code = -LTTNG_ERR_UST_CHAN_FAIL;
+       } else {
+               msg.ret_code = LTTNG_OK;
+               msg.key = channel->key;
+               msg.stream_count = channel->streams.count;
+       }
+
+       return lttcomm_send_unix_sock(sock, &msg, sizeof(msg));
+}
index 830514659772b0afa6bb5de1ad7be678e033dcfb..92f9e20957f09351834a366a38a1a708498c8dad 100644 (file)
 #include <limits.h>
 #include <poll.h>
 #include <unistd.h>
+#include <urcu/list.h>
 
 #include <lttng/lttng.h>
 
 #include <common/hashtable/hashtable.h>
 #include <common/compat/fcntl.h>
+#include <common/compat/uuid.h>
 #include <common/sessiond-comm/sessiond-comm.h>
 
 /* Commands for consumer */
@@ -43,6 +45,10 @@ enum lttng_consumer_command {
        LTTNG_CONSUMER_DESTROY_RELAYD,
        /* Return to the sessiond if there is data pending for a session */
        LTTNG_CONSUMER_DATA_PENDING,
+       /* Consumer creates a channel and returns it to sessiond. */
+       LTTNG_CONSUMER_ASK_CHANNEL_CREATION,
+       LTTNG_CONSUMER_GET_CHANNEL,
+       LTTNG_CONSUMER_DESTROY_CHANNEL,
 };
 
 /* State of each fd in consumer */
@@ -64,30 +70,60 @@ enum consumer_endpoint_status {
        CONSUMER_ENDPOINT_INACTIVE,
 };
 
+enum consumer_channel_output {
+       CONSUMER_CHANNEL_MMAP   = 0,
+       CONSUMER_CHANNEL_SPLICE = 1,
+};
+
+enum consumer_channel_type {
+       CONSUMER_CHANNEL_TYPE_METADATA  = 0,
+       CONSUMER_CHANNEL_TYPE_DATA              = 1,
+};
+
+struct stream_list {
+       struct cds_list_head head;
+       unsigned int count;
+};
+
 struct lttng_consumer_channel {
+       /* HT node used for consumer_data.channel_ht */
        struct lttng_ht_node_ulong node;
+       /* Indexed key. Incremented value in the consumer. */
        int key;
-       uint64_t max_sb_size; /* the subbuffer size for this channel */
-       int refcount; /* Number of streams referencing this channel */
+       /* Number of streams referencing this channel */
+       int refcount;
+       /* Tracing session id on the session daemon side. */
+       uint64_t session_id;
+       /* Channel trace file path name. */
+       char pathname[PATH_MAX];
+       /* Channel name. */
+       char name[LTTNG_SYMBOL_NAME_LEN];
+       /* UID and GID of the channel. */
+       uid_t uid;
+       gid_t gid;
+       /* Relayd id of the channel. -1 if it does not apply. */
+       int relayd_id;
        /*
-        * The number of streams to receive initially. Used to guarantee that we do
-        * not destroy a channel before receiving all its associated streams.
+        * Number of streams NOT initialized yet. This is used in order to not
+        * delete this channel if streams are getting initialized.
         */
-       unsigned int nb_init_streams;
+       unsigned int nb_init_stream_left;
+       /* Output type (mmap or splice). */
+       enum consumer_channel_output output;
+       /* Channel type for stream */
+       enum consumer_channel_type type;
 
        /* For UST */
-       int shm_fd;
-       int wait_fd;
-       void *mmap_base;
-       size_t mmap_len;
-       struct lttng_ust_shm_handle *handle;
-       int wait_fd_is_copy;
-       int cpucount;
+       struct ustctl_consumer_channel *uchan;
+       unsigned char uuid[UUID_STR_LEN];
+       /*
+        * Temporary stream list used to store the streams once created and waiting
+        * to be sent to the session daemon by receiving the
+        * LTTNG_CONSUMER_GET_CHANNEL.
+        */
+       struct stream_list streams;
 };
 
-/* Forward declaration for UST. */
-struct lttng_ust_lib_ring_buffer;
-
 /*
  * Internal representation of the streams, sessiond_key is used to identify
  * uniquely a stream.
@@ -97,29 +133,36 @@ struct lttng_consumer_stream {
        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 */
+       /* Pointer to associated channel. */
+       struct lttng_consumer_channel *chan;
+
+       /* Key by which the stream is indexed for 'node'. */
+       int key;
        /*
-        * key is the key used by the session daemon to refer to the
-        * object in the consumer daemon.
+        * File descriptor of the data output file. This can be either a file or a
+        * socket fd for relayd streaming.
         */
-       int key;
-       int shm_fd;
-       int wait_fd;
        int out_fd; /* output file to write the data */
-       off_t out_fd_offset; /* write position in the output file descriptor */
-       char path_name[PATH_MAX]; /* tracefile name */
+       /* Write position in the output file descriptor */
+       off_t out_fd_offset;
        enum lttng_consumer_stream_state state;
-       size_t shm_len;
-       void *mmap_base;
-       size_t mmap_len;
-       enum lttng_event_output output; /* splice or mmap */
        int shm_fd_is_copy;
-       int wait_fd_is_copy;
-       /* For UST */
-       struct lttng_ust_lib_ring_buffer *buf;
-       int cpu;
        int data_read;
        int hangup_flush_done;
+       enum lttng_event_output output;
+       /* Maximum subbuffer size. */
+       unsigned long max_sb_size;
+
+       /*
+        * Still used by the kernel for MMAP output. For UST, the ustctl getter is
+        * used for the mmap base and offset.
+        */
+       void *mmap_base;
+       unsigned long mmap_len;
+
+       /* For UST */
+
+       int wait_fd;
        /* UID/GID of the user owning the session to which stream belongs */
        uid_t uid;
        gid_t gid;
@@ -159,6 +202,11 @@ struct lttng_consumer_stream {
         * consumer data appropriate pipe.
         */
        enum consumer_endpoint_status endpoint_status;
+       /* Stream name. Format is: <channel_name>_<cpu_number> */
+       char name[LTTNG_SYMBOL_NAME_LEN];
+       /* Internal state of libustctl. */
+       struct ustctl_consumer_stream *ustream;
+       struct cds_list_head send_node;
 };
 
 /*
@@ -315,18 +363,18 @@ struct lttng_consumer_global_data {
 /*
  * Init consumer data structures.
  */
-extern void lttng_consumer_init(void);
+void lttng_consumer_init(void);
 
 /*
  * Set the error socket for communication with a session daemon.
  */
-extern void lttng_consumer_set_error_sock(
-               struct lttng_consumer_local_data *ctx, int sock);
+void lttng_consumer_set_error_sock(struct lttng_consumer_local_data *ctx,
+               int sock);
 
 /*
  * Set the command socket path for communication with a session daemon.
  */
-extern void lttng_consumer_set_command_sock_path(
+void lttng_consumer_set_command_sock_path(
                struct lttng_consumer_local_data *ctx, char *sock);
 
 /*
@@ -335,92 +383,88 @@ extern void lttng_consumer_set_command_sock_path(
  * Returns the return code of sendmsg : the number of bytes transmitted or -1
  * on error.
  */
-extern int lttng_consumer_send_error(
-               struct lttng_consumer_local_data *ctx, int cmd);
+int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, int cmd);
 
 /*
  * Called from signal handler to ensure a clean exit.
  */
-extern void lttng_consumer_should_exit(
-               struct lttng_consumer_local_data *ctx);
+void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx);
 
 /*
  * Cleanup the daemon's socket on exit.
  */
-extern void lttng_consumer_cleanup(void);
+void lttng_consumer_cleanup(void);
 
 /*
  * Flush pending writes to trace output disk file.
  */
-extern void lttng_consumer_sync_trace_file(
-               struct lttng_consumer_stream *stream, off_t orig_offset);
+void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
+               off_t orig_offset);
 
 /*
  * Poll on the should_quit pipe and the command socket return -1 on error and
  * should exit, 0 if data is available on the command socket
  */
-extern int lttng_consumer_poll_socket(struct pollfd *kconsumer_sockpoll);
+int lttng_consumer_poll_socket(struct pollfd *kconsumer_sockpoll);
 
-extern struct lttng_consumer_stream *consumer_allocate_stream(
-               int channel_key, int stream_key,
-               int shm_fd, int wait_fd,
+struct lttng_consumer_stream *consumer_allocate_stream(int channel_key,
+               int stream_key,
                enum lttng_consumer_stream_state state,
-               uint64_t mmap_len,
-               enum lttng_event_output output,
-               const char *path_name,
+               const char *channel_name,
                uid_t uid,
                gid_t gid,
-               int net_index,
-               int metadata_flag,
+               int relayd_id,
                uint64_t session_id,
-               int *alloc_ret);
-extern void consumer_del_stream(struct lttng_consumer_stream *stream,
+               int cpu,
+               int *alloc_ret,
+               enum consumer_channel_type type);
+struct lttng_consumer_channel *consumer_allocate_channel(unsigned long key,
+               uint64_t session_id,
+               const char *pathname,
+               const char *name,
+               uid_t uid,
+               gid_t gid,
+               int relayd_id,
+               enum lttng_event_output output);
+void consumer_del_stream(struct lttng_consumer_stream *stream,
                struct lttng_ht *ht);
-extern void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
+void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
                struct lttng_ht *ht);
-extern void consumer_del_channel(struct lttng_consumer_channel *channel);
-extern struct lttng_consumer_channel *consumer_allocate_channel(
-               int channel_key,
-               int shm_fd, int wait_fd,
-               uint64_t mmap_len,
-               uint64_t max_sb_size,
-               unsigned int nb_init_streams);
 int consumer_add_channel(struct lttng_consumer_channel *channel);
+void consumer_del_channel(struct lttng_consumer_channel *channel);
 
 /* lttng-relayd consumer command */
 struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
                int net_seq_idx);
 struct consumer_relayd_sock_pair *consumer_find_relayd(int key);
+struct lttng_consumer_channel *consumer_find_channel(unsigned long key);
 int consumer_handle_stream_before_relayd(struct lttng_consumer_stream *stream,
                size_t data_size);
 void consumer_steal_stream_key(int key, struct lttng_ht *ht);
 
-extern struct lttng_consumer_local_data *lttng_consumer_create(
+struct lttng_consumer_local_data *lttng_consumer_create(
                enum lttng_consumer_type type,
                ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
                        struct lttng_consumer_local_data *ctx),
                int (*recv_channel)(struct lttng_consumer_channel *channel),
                int (*recv_stream)(struct lttng_consumer_stream *stream),
                int (*update_stream)(int sessiond_key, uint32_t state));
-extern void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx);
-extern ssize_t lttng_consumer_on_read_subbuffer_mmap(
+void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx);
+ssize_t lttng_consumer_on_read_subbuffer_mmap(
                struct lttng_consumer_local_data *ctx,
                struct lttng_consumer_stream *stream, unsigned long len,
                unsigned long padding);
-extern ssize_t lttng_consumer_on_read_subbuffer_splice(
+ssize_t lttng_consumer_on_read_subbuffer_splice(
                struct lttng_consumer_local_data *ctx,
                struct lttng_consumer_stream *stream, unsigned long len,
                unsigned long padding);
-extern int lttng_consumer_take_snapshot(struct lttng_consumer_local_data *ctx,
-               struct lttng_consumer_stream *stream);
-extern int lttng_consumer_get_produced_snapshot(
-               struct lttng_consumer_local_data *ctx,
-               struct lttng_consumer_stream *stream,
+int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream);
+int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
                unsigned long *pos);
-extern void *consumer_thread_metadata_poll(void *data);
-extern void *consumer_thread_data_poll(void *data);
-extern void *consumer_thread_sessiond_poll(void *data);
-extern int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
+void *consumer_thread_metadata_poll(void *data);
+void *consumer_thread_data_poll(void *data);
+void *consumer_thread_sessiond_poll(void *data);
+int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                int sock, struct pollfd *consumer_sockpoll);
 
 ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
@@ -434,5 +478,7 @@ void consumer_flag_relayd_for_destroy(
                struct consumer_relayd_sock_pair *relayd);
 int consumer_data_pending(uint64_t id);
 int consumer_send_status_msg(int sock, int ret_code);
+int consumer_send_status_channel(int sock,
+               struct lttng_consumer_channel *channel);
 
 #endif /* LIB_CONSUMER_H */
index 0e3aecf87c7ebcd1b9def4048bb0326aa4fb82c8..93eb176a08f1f0f8e40199a8011c327850c7ae1d 100644 (file)
 #define DEFAULT_APP_SOCKET_RW_TIMEOUT       5  /* sec */
 #define DEFAULT_APP_SOCKET_TIMEOUT_ENV      "LTTNG_APP_SOCKET_TIMEOUT"
 
+#define DEFAULT_UST_STREAM_FD_NUM                      2 /* Number of fd per UST stream. */
 
 extern size_t default_channel_subbuf_size;
 extern size_t default_metadata_subbuf_size;
index 03af109a14fbc3772af12fe2661ab31a2e3d3f9a..f8c0a1d06198bf218687a0ed8e67f3036ecba5d5 100644 (file)
@@ -28,6 +28,7 @@
 #endif
 
 #include <lttng/lttng-error.h>
+#include <common/compat/tid.h>
 
 /* Stringify the expansion of a define */
 #define XSTR(d) STR(d)
@@ -36,6 +37,7 @@
 extern int lttng_opt_quiet;
 extern int lttng_opt_verbose;
 
+/* Error type. */
 #define PRINT_ERR   0x1
 #define PRINT_WARN  0x2
 #define PRINT_BUG   0x3
@@ -65,6 +67,11 @@ extern int lttng_opt_verbose;
                }                                                           \
        } while (0);
 
+/* Three level of debug. Use -v, -vv or -vvv for the levels */
+#define _ERRMSG(msg, type, fmt, args...) __lttng_print(type, msg \
+               " [%ld/%ld]: " fmt " (in %s() at " __FILE__ ":" XSTR(__LINE__) ")\n", \
+                       (long) getpid(), (long) gettid(), ## args, __func__)
+
 #define MSG(fmt, args...) \
        __lttng_print(PRINT_MSG, fmt "\n", ## args)
 #define _MSG(fmt, args...) \
@@ -73,20 +80,14 @@ extern int lttng_opt_verbose;
        __lttng_print(PRINT_ERR, "Error: " fmt "\n", ## args)
 #define WARN(fmt, args...) \
        __lttng_print(PRINT_WARN, "Warning: " fmt "\n", ## args)
-#define BUG(fmt, args...) \
-       __lttng_print(PRINT_BUG, "BUG: " fmt "\n", ## args)
 
-/* Three level of debug. Use -v, -vv or -vvv for the levels */
-#define DBG(fmt, args...) __lttng_print(PRINT_DBG, "DEBUG1: " fmt \
-               " [in %s() at " __FILE__ ":" XSTR(__LINE__) "]\n", ## args, __func__)
-#define DBG2(fmt, args...) __lttng_print(PRINT_DBG2, "DEBUG2: " fmt \
-               " [in %s() at " __FILE__ ":" XSTR(__LINE__) "]\n", ## args, __func__)
-#define DBG3(fmt, args...) __lttng_print(PRINT_DBG3, "DEBUG3: " fmt \
-               " [in %s() at " __FILE__ ":" XSTR(__LINE__) "]\n", ## args, __func__)
-
-#define _PERROR(fmt, args...) \
-       __lttng_print(PRINT_ERR, "PERROR: " fmt \
-               " [in %s() at " __FILE__ ":" XSTR(__LINE__) "]\n", ## args, __func__)
+#define BUG(fmt, args...) _ERRMSG("BUG", PRINT_BUG, fmt, ## args)
+
+#define DBG(fmt, args...) _ERRMSG("DEBUG1", PRINT_DBG, fmt, ## args)
+#define DBG2(fmt, args...) _ERRMSG("DEBUG2", PRINT_DBG2, fmt, ## args)
+#define DBG3(fmt, args...) _ERRMSG("DEBUG3", PRINT_DBG3, fmt, ## args)
+
+#define _PERROR(fmt, args...) _ERRMSG("PERROR", PRINT_ERR, fmt, ## args)
 
 #if !defined(__linux__) || ((_POSIX_C_SOURCE >= 200112L || _XOPEN_SOURCE >= 600) && !defined(_GNU_SOURCE))
 
index 21e1f41c5db3439a2a4f93584f05e58c715bea3f..9d75d3f09cfc9ada6f30503a6c7ddfe306366809 100644 (file)
@@ -47,8 +47,7 @@ extern volatile int consumer_quit;
  *
  * Returns 0 on success, < 0 on error
  */
-int lttng_kconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
-               struct lttng_consumer_stream *stream)
+int lttng_kconsumer_take_snapshot(struct lttng_consumer_stream *stream)
 {
        int ret = 0;
        int infd = stream->wait_fd;
@@ -67,9 +66,7 @@ int lttng_kconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
  *
  * Returns 0 on success, < 0 on error
  */
-int lttng_kconsumer_get_produced_snapshot(
-               struct lttng_consumer_local_data *ctx,
-               struct lttng_consumer_stream *stream,
+int lttng_kconsumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
                unsigned long *pos)
 {
        int ret;
@@ -133,14 +130,26 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 
                DBG("consumer_add_channel %d", msg.u.channel.channel_key);
                new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
-                               -1, -1,
-                               msg.u.channel.mmap_len,
-                               msg.u.channel.max_sb_size,
-                               msg.u.channel.nb_init_streams);
+                               msg.u.channel.session_id, msg.u.channel.pathname,
+                               msg.u.channel.name, msg.u.channel.uid, msg.u.channel.gid,
+                               msg.u.channel.relayd_id, msg.u.channel.output);
                if (new_channel == NULL) {
                        lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
                        goto end_nosignal;
                }
+               new_channel->nb_init_stream_left = msg.u.channel.nb_init_streams;
+
+               /* Translate and save channel type. */
+               switch (msg.u.channel.type) {
+               case CONSUMER_CHANNEL_TYPE_DATA:
+               case CONSUMER_CHANNEL_TYPE_METADATA:
+                       new_channel->type = msg.u.channel.type;
+                       break;
+               default:
+                       assert(0);
+                       goto end_nosignal;
+               };
+
                if (ctx->on_recv_channel != NULL) {
                        ret = ctx->on_recv_channel(new_channel);
                        if (ret == 0) {
@@ -158,12 +167,30 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                int fd, stream_pipe;
                struct consumer_relayd_sock_pair *relayd = NULL;
                struct lttng_consumer_stream *new_stream;
+               struct lttng_consumer_channel *channel;
                int alloc_ret = 0;
 
+               /*
+                * Get stream's channel reference. Needed when adding the stream to the
+                * global hash table.
+                */
+               channel = consumer_find_channel(msg.u.stream.channel_key);
+               if (!channel) {
+                       /*
+                        * We could not find the channel. Can happen if cpu hotplug
+                        * happens while tearing down.
+                        */
+                       ERR("Unable to find channel key %d", msg.u.stream.channel_key);
+                       ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND;
+               }
+
                /* First send a status message before receiving the fds. */
                ret = consumer_send_status_msg(sock, ret_code);
-               if (ret < 0) {
-                       /* Somehow, the session daemon is not responding anymore. */
+               if (ret < 0 || ret_code != LTTNG_OK) {
+                       /*
+                        * Somehow, the session daemon is not responding anymore or the
+                        * channel was not found.
+                        */
                        goto end_nosignal;
                }
 
@@ -192,19 +219,17 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                        goto end_nosignal;
                }
 
-               new_stream = consumer_allocate_stream(msg.u.stream.channel_key,
-                               msg.u.stream.stream_key,
-                               fd, fd,
-                               msg.u.stream.state,
-                               msg.u.stream.mmap_len,
-                               msg.u.stream.output,
-                               msg.u.stream.path_name,
-                               msg.u.stream.uid,
-                               msg.u.stream.gid,
-                               msg.u.stream.net_index,
-                               msg.u.stream.metadata_flag,
-                               msg.u.stream.session_id,
-                               &alloc_ret);
+               new_stream = consumer_allocate_stream(channel->key,
+                               fd,
+                               LTTNG_CONSUMER_ACTIVE_STREAM,
+                               channel->name,
+                               channel->uid,
+                               channel->gid,
+                               channel->relayd_id,
+                               channel->session_id,
+                               msg.u.stream.cpu,
+                               &alloc_ret,
+                               channel->type);
                if (new_stream == NULL) {
                        switch (alloc_ret) {
                        case -ENOMEM:
@@ -212,16 +237,11 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                        default:
                                lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
                                break;
-                       case -ENOENT:
-                               /*
-                                * We could not find the channel. Can happen if cpu hotplug
-                                * happens while tearing down.
-                                */
-                               DBG3("Could not find channel");
-                               break;
                        }
                        goto end_nosignal;
                }
+               new_stream->chan = channel;
+               new_stream->wait_fd = fd;
 
                /*
                 * The buffer flush is done on the session daemon side for the kernel
@@ -233,21 +253,21 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                new_stream->hangup_flush_done = 0;
 
                /* The stream is not metadata. Get relayd reference if exists. */
-               relayd = consumer_find_relayd(msg.u.stream.net_index);
+               relayd = consumer_find_relayd(new_stream->net_seq_idx);
                if (relayd != NULL) {
                        /* Add stream on the relayd */
                        pthread_mutex_lock(&relayd->ctrl_sock_mutex);
                        ret = relayd_add_stream(&relayd->control_sock,
-                                       msg.u.stream.name, msg.u.stream.path_name,
+                                       new_stream->name, new_stream->chan->pathname,
                                        &new_stream->relayd_stream_id);
                        pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
                        if (ret < 0) {
                                consumer_del_stream(new_stream, NULL);
                                goto end_nosignal;
                        }
-               } else if (msg.u.stream.net_index != -1) {
+               } else if (new_stream->net_seq_idx != -1) {
                        ERR("Network sequence index %d unknown. Not adding stream.",
-                                       msg.u.stream.net_index);
+                                       new_stream->net_seq_idx);
                        consumer_del_stream(new_stream, NULL);
                        goto end_nosignal;
                }
@@ -279,7 +299,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                }
 
                DBG("Kernel consumer ADD_STREAM %s (fd: %d) with relayd id %" PRIu64,
-                               msg.u.stream.path_name, fd, new_stream->relayd_stream_id);
+                               new_stream->name, fd, new_stream->relayd_stream_id);
                break;
        }
        case LTTNG_CONSUMER_UPDATE_STREAM:
@@ -394,7 +414,7 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
                goto end;
        }
 
-       switch (stream->output) {
+       switch (stream->chan->output) {
        case LTTNG_EVENT_SPLICE:
 
                /*
@@ -481,16 +501,23 @@ end:
 int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
 {
        int ret;
+       char full_path[PATH_MAX];
+
+       assert(stream);
+
+       ret = snprintf(full_path, sizeof(full_path), "%s/%s",
+                       stream->chan->pathname, stream->name);
+       if (ret < 0) {
+               PERROR("snprintf on_recv_stream");
+               goto error;
+       }
 
        /* Opening the tracefile in write mode */
-       if (strlen(stream->path_name) > 0 && stream->net_seq_idx == -1) {
-               ret = run_as_open(stream->path_name,
-                               O_WRONLY|O_CREAT|O_TRUNC,
-                               S_IRWXU|S_IRWXG|S_IRWXO,
-                               stream->uid, stream->gid);
+       if (stream->net_seq_idx == -1) {
+               ret = run_as_open(full_path, O_WRONLY | O_CREAT | O_TRUNC,
+                               S_IRWXU|S_IRWXG|S_IRWXO, stream->uid, stream->gid);
                if (ret < 0) {
-                       ERR("Opening %s", stream->path_name);
-                       perror("open");
+                       PERROR("open kernel stream path %s", full_path);
                        goto error;
                }
                stream->out_fd = ret;
@@ -503,15 +530,15 @@ int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
                ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len);
                if (ret != 0) {
                        errno = -ret;
-                       perror("kernctl_get_mmap_len");
+                       PERROR("kernctl_get_mmap_len");
                        goto error_close_fd;
                }
                stream->mmap_len = (size_t) mmap_len;
 
-               stream->mmap_base = mmap(NULL, stream->mmap_len,
-                               PROT_READ, MAP_PRIVATE, stream->wait_fd, 0);
+               stream->mmap_base = mmap(NULL, stream->mmap_len, PROT_READ,
+                               MAP_PRIVATE, stream->wait_fd, 0);
                if (stream->mmap_base == MAP_FAILED) {
-                       perror("Error mmaping");
+                       PERROR("Error mmaping");
                        ret = -1;
                        goto error_close_fd;
                }
index 8ae1834020f1a6d2cb7701bd9f45bc9a7423a203..41482b13f4cea8d4763515102643d6dad756a611 100644 (file)
 
 #include <common/consumer.h>
 
-/*
- * Take a snapshot for a specific fd
- *
- * Returns 0 on success, < 0 on error
- */
-int lttng_kconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
-        struct lttng_consumer_stream *stream);
-
-/*
- * Get the produced position
- *
- * Returns 0 on success, < 0 on error
- */
-int lttng_kconsumer_get_produced_snapshot(
-        struct lttng_consumer_local_data *ctx,
-        struct lttng_consumer_stream *stream,
+int lttng_kconsumer_take_snapshot(struct lttng_consumer_stream *stream);
+int lttng_kconsumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
         unsigned long *pos);
-
 int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                int sock, struct pollfd *consumer_sockpoll);
-
-
 ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
                struct lttng_consumer_local_data *ctx);
 int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream);
index 63a55c22756fd8737c00cb0158bdbb5a01b7c007..9e10c83e02f7ec583bff72cd5b6d70b42870dca1 100644 (file)
@@ -327,15 +327,44 @@ int relayd_connect(struct lttcomm_sock *sock)
 
 /*
  * Close relayd socket with an allocated lttcomm_sock.
+ *
+ * If no socket operations are found, simply return 0 meaning that everything
+ * is fine. Without operations, the socket can not possibly be opened or used.
+ * This is possible if the socket was allocated but not created. However, the
+ * caller could simply use it to store a valid file descriptor for instance
+ * passed over a Unix socket and call this to cleanup but still without a valid
+ * ops pointer.
+ *
+ * Return the close returned value. On error, a negative value is usually
+ * returned back from close(2).
  */
 int relayd_close(struct lttcomm_sock *sock)
 {
+       int ret;
+
        /* Code flow error. Safety net. */
        assert(sock);
 
+       /* An invalid fd is fine, return success. */
+       if (sock->fd < 0) {
+               ret = 0;
+               goto end;
+       }
+
        DBG3("Relayd closing socket %d", sock->fd);
 
-       return sock->ops->close(sock);
+       if (sock->ops) {
+               ret = sock->ops->close(sock);
+       } else {
+               /* Default call if no specific ops found. */
+               ret = close(sock->fd);
+               if (ret < 0) {
+                       PERROR("relayd_close default close");
+               }
+       }
+
+end:
+       return ret;
 }
 
 /*
index 322400bb2cb7ce91f6a2194250c6e875aca7061d..c390a9f5284a573a49f0c524750f5d7c66eaf25f 100644 (file)
@@ -31,6 +31,7 @@
 #include <common/compat/socket.h>
 #include <common/uri.h>
 #include <common/defaults.h>
+#include <common/compat/uuid.h>
 
 #include <arpa/inet.h>
 #include <netinet/in.h>
@@ -115,6 +116,7 @@ enum lttcomm_return_code {
        LTTCOMM_CONSUMERD_SPLICE_EINVAL,            /* EINVAL from splice(2) */
        LTTCOMM_CONSUMERD_SPLICE_ENOMEM,            /* ENOMEM from splice(2) */
        LTTCOMM_CONSUMERD_SPLICE_ESPIPE,            /* ESPIPE from splice(2) */
+       LTTCOMM_CONSUMERD_ENOMEM,                   /* Consumer is out of memory */
 
        /* MUST be last element */
        LTTCOMM_NR,                                             /* Last element */
@@ -258,28 +260,23 @@ struct lttcomm_consumer_msg {
        union {
                struct {
                        int channel_key;
-                       uint64_t max_sb_size; /* the subbuffer size for this channel */
-                       /* shm_fd and wait_fd are sent as ancillary data */
-                       uint64_t mmap_len;
+                       uint64_t session_id;
+                       char pathname[PATH_MAX];
+                       uid_t uid;
+                       gid_t gid;
+                       int relayd_id;
                        /* nb_init_streams is the number of streams open initially. */
                        unsigned int nb_init_streams;
                        char name[LTTNG_SYMBOL_NAME_LEN];
-               } LTTNG_PACKED channel;
+                       /* Use splice or mmap to consume this fd */
+                       enum lttng_event_output output;
+                       int type; /* Per cpu or metadata. */
+               } LTTNG_PACKED channel; /* Only used by Kernel. */
                struct {
-                       int channel_key;
                        int stream_key;
-                       /* shm_fd and wait_fd are sent as ancillary data */
-                       uint32_t state;    /* enum lttcomm_consumer_fd_state */
-                       enum lttng_event_output output; /* use splice or mmap to consume this fd */
-                       uint64_t mmap_len;
-                       uid_t uid;         /* User ID owning the session */
-                       gid_t gid;         /* Group ID owning the session */
-                       char path_name[PATH_MAX];
-                       int net_index;
-                       unsigned int metadata_flag;
-                       char name[DEFAULT_STREAM_NAME_LEN];  /* Name string of the stream */
-                       uint64_t session_id;   /* Tracing session id of the stream */
-               } LTTNG_PACKED stream;
+                       int channel_key;
+                       int cpu;        /* On which CPU this stream is assigned. */
+               } LTTNG_PACKED stream;  /* Only used by Kernel. */
                struct {
                        int net_index;
                        enum lttng_stream_type type;
@@ -294,6 +291,29 @@ struct lttcomm_consumer_msg {
                struct {
                        uint64_t session_id;
                } LTTNG_PACKED data_pending;
+               struct {
+                       uint64_t subbuf_size;                           /* bytes */
+                       uint64_t num_subbuf;                            /* power of 2 */
+                       int overwrite;                                          /* 1: overwrite, 0: discard */
+                       unsigned int switch_timer_interval;     /* usec */
+                       unsigned int read_timer_interval;       /* usec */
+                       int output;                                                     /* splice, mmap */
+                       int type;                                                       /* metadata or per_cpu */
+                       uint64_t session_id;                            /* Tracing session id */
+                       char pathname[PATH_MAX];                        /* Channel file path. */
+                       char name[LTTNG_SYMBOL_NAME_LEN];       /* Channel name. */
+                       uid_t uid;                                                      /* User ID of the session */
+                       gid_t gid;                                                      /* Group ID ot the session */
+                       int relayd_id;                                          /* Relayd id if apply. */
+                       unsigned long key;                                      /* Unique channel key. */
+                       unsigned char uuid[UUID_STR_LEN];       /* uuid for ust tracer. */
+               } LTTNG_PACKED ask_channel;
+               struct {
+                       unsigned long key;
+               } LTTNG_PACKED get_channel;
+               struct {
+                       unsigned long key;
+               } LTTNG_PACKED destroy_channel;
        } u;
 } LTTNG_PACKED;
 
@@ -304,6 +324,12 @@ struct lttcomm_consumer_status_msg {
        enum lttng_error_code ret_code;
 } LTTNG_PACKED;
 
+struct lttcomm_consumer_status_channel {
+       enum lttng_error_code ret_code;
+       unsigned long key;
+       unsigned int stream_count;
+} LTTNG_PACKED;
+
 #ifdef HAVE_LIBLTTNG_UST_CTL
 
 #include <lttng/ust-abi.h>
index 5dbd6ba4b4aff3feea113a2092165d13bf5bd160..442925754c634ec4d067382b5cdd7ed8f35b385e 100644 (file)
@@ -29,6 +29,7 @@
 #include <sys/types.h>
 #include <inttypes.h>
 #include <unistd.h>
+#include <urcu/list.h>
 
 #include <common/common.h>
 #include <common/sessiond-comm/sessiond-comm.h>
@@ -42,53 +43,432 @@ extern int consumer_poll_timeout;
 extern volatile int consumer_quit;
 
 /*
- * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
- * compiled out, we isolate it in this library.
+ * Free channel object and all streams associated with it. This MUST be used
+ * only and only if the channel has _NEVER_ been added to the global channel
+ * hash table.
  */
-int lttng_ustctl_get_mmap_read_offset(struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf, unsigned long *off)
+static void destroy_channel(struct lttng_consumer_channel *channel)
 {
-       return ustctl_get_mmap_read_offset(handle, buf, off);
-};
+       struct lttng_consumer_stream *stream, *stmp;
+
+       assert(channel);
+
+       DBG("UST consumer cleaning stream list");
+
+       cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
+                       send_node) {
+               cds_list_del(&stream->send_node);
+               ustctl_destroy_stream(stream->ustream);
+               free(stream);
+       }
+
+       /*
+        * If a channel is available meaning that was created before the streams
+        * were, delete it.
+        */
+       if (channel->uchan) {
+               lttng_ustconsumer_del_channel(channel);
+       }
+       free(channel);
+}
 
 /*
- * Take a snapshot for a specific fd
+ * Add channel to internal consumer state.
  *
- * Returns 0 on success, < 0 on error
+ * Returns 0 on success or else a negative value.
  */
-int lttng_ustconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
-               struct lttng_consumer_stream *stream)
+static int add_channel(struct lttng_consumer_channel *channel,
+               struct lttng_consumer_local_data *ctx)
 {
        int ret = 0;
 
-       ret = ustctl_snapshot(stream->chan->handle, stream->buf);
-       if (ret != 0) {
-               errno = -ret;
-               PERROR("Getting sub-buffer snapshot.");
+       assert(channel);
+       assert(ctx);
+
+       if (ctx->on_recv_channel != NULL) {
+               ret = ctx->on_recv_channel(channel);
+               if (ret == 0) {
+                       ret = consumer_add_channel(channel);
+               } else if (ret < 0) {
+                       /* Most likely an ENOMEM. */
+                       lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
+                       goto error;
+               }
+       } else {
+               ret = consumer_add_channel(channel);
        }
 
+       DBG("UST consumer channel added (key: %u)", channel->key);
+
+error:
        return ret;
 }
 
 /*
- * Get the produced position
+ * Allocate and return a consumer channel object.
+ */
+static struct lttng_consumer_channel *allocate_channel(uint64_t session_id,
+               const char *pathname, const char *name, uid_t uid, gid_t gid,
+               int relayd_id, unsigned long key, enum lttng_event_output output)
+{
+       assert(pathname);
+       assert(name);
+
+       return consumer_allocate_channel(key, session_id, pathname, name, uid, gid,
+                       relayd_id, output);
+}
+
+/*
+ * Allocate and return a consumer stream object. If _alloc_ret is not NULL, the
+ * error value if applicable is set in it else it is kept untouched.
  *
- * Returns 0 on success, < 0 on error
+ * Return NULL on error else the newly allocated stream object.
  */
-int lttng_ustconsumer_get_produced_snapshot(
-               struct lttng_consumer_local_data *ctx,
-               struct lttng_consumer_stream *stream,
-               unsigned long *pos)
+static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
+               struct lttng_consumer_channel *channel,
+               struct lttng_consumer_local_data *ctx, int *_alloc_ret)
+{
+       int alloc_ret;
+       struct lttng_consumer_stream *stream = NULL;
+
+       assert(channel);
+       assert(ctx);
+
+       stream = consumer_allocate_stream(channel->key,
+                       key,
+                       LTTNG_CONSUMER_ACTIVE_STREAM,
+                       channel->name,
+                       channel->uid,
+                       channel->gid,
+                       channel->relayd_id,
+                       channel->session_id,
+                       cpu,
+                       &alloc_ret,
+                       channel->type);
+       if (stream == NULL) {
+               switch (alloc_ret) {
+               case -ENOENT:
+                       /*
+                        * We could not find the channel. Can happen if cpu hotplug
+                        * happens while tearing down.
+                        */
+                       DBG3("Could not find channel");
+                       break;
+               case -ENOMEM:
+               case -EINVAL:
+               default:
+                       lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
+                       break;
+               }
+               goto error;
+       }
+
+       stream->chan = channel;
+
+error:
+       if (_alloc_ret) {
+               *_alloc_ret = alloc_ret;
+       }
+       return stream;
+}
+
+/*
+ * Send the given stream pointer to the corresponding thread.
+ *
+ * Returns 0 on success else a negative value.
+ */
+static int send_stream_to_thread(struct lttng_consumer_stream *stream,
+               struct lttng_consumer_local_data *ctx)
+{
+       int ret, stream_pipe;
+
+       /* Get the right pipe where the stream will be sent. */
+       if (stream->metadata_flag) {
+               stream_pipe = ctx->consumer_metadata_pipe[1];
+       } else {
+               stream_pipe = ctx->consumer_data_pipe[1];
+       }
+
+       do {
+               ret = write(stream_pipe, &stream, sizeof(stream));
+       } while (ret < 0 && errno == EINTR);
+       if (ret < 0) {
+               PERROR("Consumer write %s stream to pipe %d",
+                               stream->metadata_flag ? "metadata" : "data", stream_pipe);
+       }
+
+       return ret;
+}
+
+/*
+ * Search for a relayd object related to the stream. If found, send the stream
+ * to the relayd.
+ *
+ * On success, returns 0 else a negative value.
+ */
+static int send_stream_to_relayd(struct lttng_consumer_stream *stream)
+{
+       int ret = 0;
+       struct consumer_relayd_sock_pair *relayd;
+
+       assert(stream);
+
+       relayd = consumer_find_relayd(stream->net_seq_idx);
+       if (relayd != NULL) {
+               pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+               /* Add stream on the relayd */
+               ret = relayd_add_stream(&relayd->control_sock, stream->name,
+                               stream->chan->pathname, &stream->relayd_stream_id);
+               pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+               if (ret < 0) {
+                       goto error;
+               }
+       } else if (stream->net_seq_idx != -1) {
+               ERR("Network sequence index %d unknown. Not adding stream.",
+                               stream->net_seq_idx);
+               ret = -1;
+               goto error;
+       }
+
+error:
+       return ret;
+}
+
+static int create_ust_streams(struct lttng_consumer_channel *channel,
+               struct lttng_consumer_local_data *ctx)
+{
+       int ret, cpu = 0;
+       struct ustctl_consumer_stream *ustream;
+       struct lttng_consumer_stream *stream;
+
+       assert(channel);
+       assert(ctx);
+
+       /*
+        * While a stream is available from ustctl. When NULL is returned, we've
+        * reached the end of the possible stream for the channel.
+        */
+       while ((ustream = ustctl_create_stream(channel->uchan, cpu))) {
+               int wait_fd;
+
+               wait_fd = ustctl_get_wait_fd(ustream);
+
+               /* Allocate consumer stream object. */
+               stream = allocate_stream(cpu, wait_fd, channel, ctx, &ret);
+               if (!stream) {
+                       goto error_alloc;
+               }
+               stream->ustream = ustream;
+               /*
+                * Store it so we can save multiple function calls afterwards since
+                * this value is used heavily in the stream threads. This is UST
+                * specific so this is why it's done after allocation.
+                */
+               stream->wait_fd = wait_fd;
+
+               /*
+                * Order is important this is why a list is used. On error, the caller
+                * should clean this list.
+                */
+               cds_list_add_tail(&stream->send_node, &channel->streams.head);
+
+               ret = ustctl_get_max_subbuf_size(stream->ustream,
+                               &stream->max_sb_size);
+               if (ret < 0) {
+                       ERR("ustctl_get_max_subbuf_size failed for stream %s",
+                                       stream->name);
+                       goto error;
+               }
+
+               /* Do actions once stream has been received. */
+               if (ctx->on_recv_stream) {
+                       ret = ctx->on_recv_stream(stream);
+                       if (ret < 0) {
+                               goto error;
+                       }
+               }
+
+               DBG("UST consumer add stream %s (key: %d) with relayd id %" PRIu64,
+                               stream->name, stream->key, stream->relayd_stream_id);
+
+               /* Set next CPU stream. */
+               channel->streams.count = ++cpu;
+       }
+
+       return 0;
+
+error:
+error_alloc:
+       return ret;
+}
+
+/*
+ * Create an UST channel with the given attributes and send it to the session
+ * daemon using the ust ctl API.
+ *
+ * Return 0 on success or else a negative value.
+ */
+static int create_ust_channel(struct ustctl_consumer_channel_attr *attr,
+               struct ustctl_consumer_channel **chanp)
+{
+       int ret;
+       struct ustctl_consumer_channel *channel;
+
+       assert(attr);
+       assert(chanp);
+
+       DBG3("Creating channel to ustctl with attr: [overwrite: %d, "
+                       "subbuf_size: %" PRIu64 ", num_subbuf: %" PRIu64 ", "
+                       "switch_timer_interval: %u, read_timer_interval: %u, "
+                       "output: %d, type: %d", attr->overwrite, attr->subbuf_size,
+                       attr->num_subbuf, attr->switch_timer_interval,
+                       attr->read_timer_interval, attr->output, attr->type);
+
+       channel = ustctl_create_channel(attr);
+       if (!channel) {
+               ret = -1;
+               goto error_create;
+       }
+
+       *chanp = channel;
+
+       return 0;
+
+error_create:
+       return ret;
+}
+
+static int send_sessiond_stream(int sock, struct lttng_consumer_stream *stream)
+{
+       int ret;
+
+       assert(stream);
+       assert(sock >= 0);
+
+       DBG2("UST consumer sending stream %d to sessiond", stream->key);
+
+       /* Send stream to session daemon. */
+       ret = ustctl_send_stream_to_sessiond(sock, stream->ustream);
+       if (ret < 0) {
+               goto error;
+       }
+
+       ret = ustctl_stream_close_wakeup_fd(stream->ustream);
+       if (ret < 0) {
+               goto error;
+       }
+
+error:
+       return ret;
+}
+
+/*
+ * Send channel to sessiond.
+ *
+ * Return 0 on success or else a negative value. On error, the channel is
+ * destroy using ustctl.
+ */
+static int send_sessiond_channel(int sock,
+               struct lttng_consumer_channel *channel,
+               struct lttng_consumer_local_data *ctx, int *relayd_error)
+{
+       int ret;
+       struct lttng_consumer_stream *stream;
+
+       assert(channel);
+       assert(ctx);
+       assert(sock >= 0);
+
+       DBG("UST consumer sending channel %s to sessiond", channel->name);
+
+       /* Send channel to sessiond. */
+       ret = ustctl_send_channel_to_sessiond(sock, channel->uchan);
+       if (ret < 0) {
+               goto error;
+       }
+
+       /* The channel was sent successfully to the sessiond at this point. */
+       cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
+               /* Try to send the stream to the relayd if one is available. */
+               ret = send_stream_to_relayd(stream);
+               if (ret < 0) {
+                       /*
+                        * Flag that the relayd was the problem here probably due to a
+                        * communicaton error on the socket.
+                        */
+                       if (relayd_error) {
+                               *relayd_error = 1;
+                       }
+                       goto error;
+               }
+
+               /* Send stream to session daemon. */
+               ret = send_sessiond_stream(sock, stream);
+               if (ret < 0) {
+                       goto error;
+               }
+       }
+
+       /* Tell sessiond there is no more stream. */
+       ret = ustctl_send_stream_to_sessiond(sock, NULL);
+       if (ret < 0) {
+               goto error;
+       }
+
+       DBG("UST consumer NULL stream sent to sessiond");
+
+       return 0;
+
+error:
+       return ret;
+}
+
+/*
+ * Creates a channel and streams and add the channel it to the channel internal
+ * state. The created stream must ONLY be sent once the GET_CHANNEL command is
+ * received.
+ *
+ * Return 0 on success or else, a negative value is returned and the channel
+ * MUST be destroyed by consumer_del_channel().
+ */
+static int ask_channel(struct lttng_consumer_local_data *ctx, int sock,
+               struct lttng_consumer_channel *channel,
+               struct ustctl_consumer_channel_attr *attr)
 {
        int ret;
 
-       ret = ustctl_snapshot_get_produced(stream->chan->handle,
-                       stream->buf, pos);
-       if (ret != 0) {
-               errno = -ret;
-               PERROR("ustctl_snapshot_get_produced");
+       assert(ctx);
+       assert(channel);
+       assert(attr);
+
+       /*
+        * This value is still used by the kernel consumer since for the kernel,
+        * the stream ownership is not IN the consumer so we need to have the
+        * number of left stream that needs to be initialized so we can know when
+        * to delete the channel (see consumer.c).
+        *
+        * As for the user space tracer now, the consumer creates and sends the
+        * stream to the session daemon which only sends them to the application
+        * once every stream of a channel is received making this value useless
+        * because we they will be added to the poll thread before the application
+        * receives them. This ensures that a stream can not hang up during
+        * initilization of a channel.
+        */
+       channel->nb_init_stream_left = 0;
+
+       /* The reply msg status is handled in the following call. */
+       ret = create_ust_channel(attr, &channel->uchan);
+       if (ret < 0) {
+               goto error;
        }
 
+       /* Open all streams for this channel. */
+       ret = create_ust_streams(channel, ctx);
+       if (ret < 0) {
+               goto error;
+       }
+
+error:
        return ret;
 }
 
@@ -103,12 +483,13 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
        ssize_t ret;
        enum lttng_error_code ret_code = LTTNG_OK;
        struct lttcomm_consumer_msg msg;
+       struct lttng_consumer_channel *channel = NULL;
 
        ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
        if (ret != sizeof(msg)) {
                DBG("Consumer received unexpected message size %zd (expects %zu)",
                        ret, sizeof(msg));
-               lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
+               lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
                /*
                 * The ret value might 0 meaning an orderly shutdown but this is ok
                 * since the caller handles this.
@@ -139,202 +520,6 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                                &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id);
                goto end_nosignal;
        }
-       case LTTNG_CONSUMER_ADD_CHANNEL:
-       {
-               struct lttng_consumer_channel *new_channel;
-               int fds[1];
-               size_t nb_fd = 1;
-
-               DBG("UST Consumer adding channel");
-
-               /* First send a status message before receiving the fds. */
-               ret = consumer_send_status_msg(sock, ret_code);
-               if (ret < 0) {
-                       /* Somehow, the session daemon is not responding anymore. */
-                       goto end_nosignal;
-               }
-
-               /* block */
-               if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
-                       rcu_read_unlock();
-                       return -EINTR;
-               }
-               ret = lttcomm_recv_fds_unix_sock(sock, fds, nb_fd);
-               if (ret != sizeof(fds)) {
-                       lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
-                       rcu_read_unlock();
-                       /*
-                        * The ret value might 0 meaning an orderly shutdown but this is ok
-                        * since the caller handles this.
-                        */
-                       return ret;
-               }
-
-               /*
-                * Send status code to session daemon only if the recv works. If the
-                * above recv() failed, the session daemon is notified through the
-                * error socket and the teardown is eventually done.
-                */
-               ret = consumer_send_status_msg(sock, ret_code);
-               if (ret < 0) {
-                       /* Somehow, the session daemon is not responding anymore. */
-                       goto end_nosignal;
-               }
-
-               DBG("consumer_add_channel %d", msg.u.channel.channel_key);
-
-               new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
-                               fds[0], -1,
-                               msg.u.channel.mmap_len,
-                               msg.u.channel.max_sb_size,
-                               msg.u.channel.nb_init_streams);
-               if (new_channel == NULL) {
-                       lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
-                       goto end_nosignal;
-               }
-               if (ctx->on_recv_channel != NULL) {
-                       ret = ctx->on_recv_channel(new_channel);
-                       if (ret == 0) {
-                               consumer_add_channel(new_channel);
-                       } else if (ret < 0) {
-                               goto end_nosignal;
-                       }
-               } else {
-                       consumer_add_channel(new_channel);
-               }
-               goto end_nosignal;
-       }
-       case LTTNG_CONSUMER_ADD_STREAM:
-       {
-               struct lttng_consumer_stream *new_stream;
-               int fds[2], stream_pipe;
-               size_t nb_fd = 2;
-               struct consumer_relayd_sock_pair *relayd = NULL;
-               int alloc_ret = 0;
-
-               DBG("UST Consumer adding stream");
-
-               /* First send a status message before receiving the fds. */
-               ret = consumer_send_status_msg(sock, ret_code);
-               if (ret < 0) {
-                       /* Somehow, the session daemon is not responding anymore. */
-                       goto end_nosignal;
-               }
-
-               /* block */
-               if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
-                       rcu_read_unlock();
-                       return -EINTR;
-               }
-               ret = lttcomm_recv_fds_unix_sock(sock, fds, nb_fd);
-               if (ret != sizeof(fds)) {
-                       lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
-                       rcu_read_unlock();
-                       /*
-                        * The ret value might 0 meaning an orderly shutdown but this is ok
-                        * since the caller handles this.
-                        */
-                       return ret;
-               }
-
-               /*
-                * Send status code to session daemon only if the recv works. If the
-                * above recv() failed, the session daemon is notified through the
-                * error socket and the teardown is eventually done.
-                */
-               ret = consumer_send_status_msg(sock, ret_code);
-               if (ret < 0) {
-                       /* Somehow, the session daemon is not responding anymore. */
-                       goto end_nosignal;
-               }
-
-               DBG("Consumer command ADD_STREAM chan %d stream %d",
-                               msg.u.stream.channel_key, msg.u.stream.stream_key);
-
-               assert(msg.u.stream.output == LTTNG_EVENT_MMAP);
-               new_stream = consumer_allocate_stream(msg.u.stream.channel_key,
-                               msg.u.stream.stream_key,
-                               fds[0], fds[1],
-                               msg.u.stream.state,
-                               msg.u.stream.mmap_len,
-                               msg.u.stream.output,
-                               msg.u.stream.path_name,
-                               msg.u.stream.uid,
-                               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) {
-                       case -ENOMEM:
-                       case -EINVAL:
-                       default:
-                               lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
-                               break;
-                       case -ENOENT:
-                               /*
-                                * We could not find the channel. Can happen if cpu hotplug
-                                * happens while tearing down.
-                                */
-                               DBG3("Could not find channel");
-                               break;
-                       }
-                       goto end_nosignal;
-               }
-
-               /* The stream is not metadata. Get relayd reference if exists. */
-               relayd = consumer_find_relayd(msg.u.stream.net_index);
-               if (relayd != NULL) {
-                       pthread_mutex_lock(&relayd->ctrl_sock_mutex);
-                       /* Add stream on the relayd */
-                       ret = relayd_add_stream(&relayd->control_sock,
-                                       msg.u.stream.name, msg.u.stream.path_name,
-                                       &new_stream->relayd_stream_id);
-                       pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
-                       if (ret < 0) {
-                               consumer_del_stream(new_stream, NULL);
-                               goto end_nosignal;
-                       }
-               } else if (msg.u.stream.net_index != -1) {
-                       ERR("Network sequence index %d unknown. Not adding stream.",
-                                       msg.u.stream.net_index);
-                       consumer_del_stream(new_stream, NULL);
-                       goto end_nosignal;
-               }
-
-               /* Do actions once stream has been received. */
-               if (ctx->on_recv_stream) {
-                       ret = ctx->on_recv_stream(new_stream);
-                       if (ret < 0) {
-                               consumer_del_stream(new_stream, NULL);
-                               goto end_nosignal;
-                       }
-               }
-
-               /* Get the right pipe where the stream will be sent. */
-               if (new_stream->metadata_flag) {
-                       stream_pipe = ctx->consumer_metadata_pipe[1];
-               } else {
-                       stream_pipe = ctx->consumer_data_pipe[1];
-               }
-
-               do {
-                       ret = write(stream_pipe, &new_stream, sizeof(new_stream));
-               } while (ret < 0 && errno == EINTR);
-               if (ret < 0) {
-                       PERROR("Consumer write %s stream to pipe %d",
-                                       new_stream->metadata_flag ? "metadata" : "data",
-                                       stream_pipe);
-                       consumer_del_stream(new_stream, NULL);
-                       goto end_nosignal;
-               }
-
-               DBG("UST consumer ADD_STREAM %s (%d,%d) with relayd id %" PRIu64,
-                               msg.u.stream.path_name, fds[0], fds[1],
-                               new_stream->relayd_stream_id);
-               break;
-       }
        case LTTNG_CONSUMER_DESTROY_RELAYD:
        {
                uint64_t index = msg.u.destroy_relayd.net_seq_idx;
@@ -398,6 +583,178 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                 */
                break;
        }
+       case LTTNG_CONSUMER_ASK_CHANNEL_CREATION:
+       {
+               int ret;
+               struct ustctl_consumer_channel_attr attr;
+
+               /* Create a plain object and reserve a channel key. */
+               channel = allocate_channel(msg.u.ask_channel.session_id,
+                               msg.u.ask_channel.pathname, msg.u.ask_channel.name,
+                               msg.u.ask_channel.uid, msg.u.ask_channel.gid,
+                               msg.u.ask_channel.relayd_id, msg.u.ask_channel.key,
+                               (enum lttng_event_output) msg.u.ask_channel.output);
+               if (!channel) {
+                       goto end_channel_error;
+               }
+
+               /* Build channel attributes from received message. */
+               attr.subbuf_size = msg.u.ask_channel.subbuf_size;
+               attr.num_subbuf = msg.u.ask_channel.num_subbuf;
+               attr.overwrite = msg.u.ask_channel.overwrite;
+               attr.switch_timer_interval = msg.u.ask_channel.switch_timer_interval;
+               attr.read_timer_interval = msg.u.ask_channel.read_timer_interval;
+               memcpy(attr.uuid, msg.u.ask_channel.uuid, sizeof(attr.uuid));
+
+               /* Translate the event output type to UST. */
+               switch (channel->output) {
+               case LTTNG_EVENT_SPLICE:
+                       /* Splice not supported so fallback on mmap(). */
+               case LTTNG_EVENT_MMAP:
+               default:
+                       attr.output = CONSUMER_CHANNEL_MMAP;
+                       break;
+               };
+
+               /* Translate and save channel type. */
+               switch (msg.u.ask_channel.type) {
+               case LTTNG_UST_CHAN_PER_CPU:
+                       channel->type = CONSUMER_CHANNEL_TYPE_DATA;
+                       attr.type = LTTNG_UST_CHAN_PER_CPU;
+                       break;
+               case LTTNG_UST_CHAN_METADATA:
+                       channel->type = CONSUMER_CHANNEL_TYPE_METADATA;
+                       attr.type = LTTNG_UST_CHAN_METADATA;
+                       break;
+               default:
+                       assert(0);
+                       goto error_fatal;
+               };
+
+               ret = ask_channel(ctx, sock, channel, &attr);
+               if (ret < 0) {
+                       goto end_channel_error;
+               }
+
+               /*
+                * Add the channel to the internal state AFTER all streams were created
+                * and successfully sent to session daemon. This way, all streams must
+                * be ready before this channel is visible to the threads.
+                */
+               ret = add_channel(channel, ctx);
+               if (ret < 0) {
+                       goto end_channel_error;
+               }
+
+               /*
+                * Channel and streams are now created. Inform the session daemon that
+                * everything went well and should wait to receive the channel and
+                * streams with ustctl API.
+                */
+               ret = consumer_send_status_channel(sock, channel);
+               if (ret < 0) {
+                       /*
+                        * There is probably a problem on the socket so the poll will get
+                        * it and clean everything up.
+                        */
+                       goto end_nosignal;
+               }
+
+               break;
+       }
+       case LTTNG_CONSUMER_GET_CHANNEL:
+       {
+               int ret, relayd_err = 0;
+               unsigned long key = msg.u.get_channel.key;
+               struct lttng_consumer_channel *channel;
+               struct lttng_consumer_stream *stream, *stmp;
+
+               channel = consumer_find_channel(key);
+               if (!channel) {
+                       ERR("UST consumer get channel key %lu not found", key);
+                       ret_code = LTTNG_ERR_UST_CHAN_NOT_FOUND;
+                       goto end_msg_sessiond;
+               }
+
+               /* Inform sessiond that we are about to send channel and streams. */
+               ret = consumer_send_status_msg(sock, LTTNG_OK);
+               if (ret < 0) {
+                       /* Somehow, the session daemon is not responding anymore. */
+                       goto end_nosignal;
+               }
+
+               /* Send everything to sessiond. */
+               ret = send_sessiond_channel(sock, channel, ctx, &relayd_err);
+               if (ret < 0) {
+                       if (relayd_err) {
+                               /*
+                                * We were unable to send to the relayd the stream so avoid
+                                * sending back a fatal error to the thread since this is OK
+                                * and the consumer can continue its work.
+                                */
+                               ret_code = LTTNG_ERR_RELAYD_CONNECT_FAIL;
+                               goto end_msg_sessiond;
+                       }
+                       /*
+                        * The communicaton was broken hence there is a bad state between
+                        * the consumer and sessiond so stop everything.
+                        */
+                       goto error_fatal;
+               }
+
+               /* Send streams to the corresponding thread. */
+               cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
+                               send_node) {
+                       /* Sending the stream to the thread. */
+                       ret = send_stream_to_thread(stream, ctx);
+                       if (ret < 0) {
+                               /*
+                                * If we are unable to send the stream to the thread, there is
+                                * a big problem so just stop everything.
+                                */
+                               goto error_fatal;
+                       }
+
+                       /* Remove node from the channel stream list. */
+                       cds_list_del(&stream->send_node);
+               }
+
+               /* List MUST be empty after or else it could be reused. */
+               assert(cds_list_empty(&channel->streams.head));
+
+               /* Inform sessiond that everything is done and OK on our side. */
+               ret = consumer_send_status_msg(sock, LTTNG_OK);
+               if (ret < 0) {
+                       /* Somehow, the session daemon is not responding anymore. */
+                       goto end_nosignal;
+               }
+
+               break;
+       }
+       case LTTNG_CONSUMER_DESTROY_CHANNEL:
+       {
+               int ret;
+               unsigned long key = msg.u.destroy_channel.key;
+               struct lttng_consumer_channel *channel;
+
+               DBG("UST consumer destroy channel key %lu", key);
+
+               channel = consumer_find_channel(key);
+               if (!channel) {
+                       ERR("UST consumer destroy channel %lu not found", key);
+                       ret_code = LTTNG_ERR_UST_CHAN_NOT_FOUND;
+               } else {
+                       /* Protocol error if the stream list is NOT empty. */
+                       assert(!cds_list_empty(&channel->streams.head));
+                       consumer_del_channel(channel);
+               }
+
+               ret = consumer_send_status_msg(sock, LTTNG_OK);
+               if (ret < 0) {
+                       /* Somehow, the session daemon is not responding anymore. */
+                       goto end_nosignal;
+               }
+       }
        default:
                break;
        }
@@ -410,83 +767,118 @@ end_nosignal:
         * shutdown during the recv() or send() call.
         */
        return 1;
+
+end_msg_sessiond:
+       /*
+        * The returned value here is not useful since either way we'll return 1 to
+        * the caller because the session daemon socket management is done
+        * elsewhere. Returning a negative code or 0 will shutdown the consumer.
+        */
+       (void) consumer_send_status_msg(sock, ret_code);
+       rcu_read_unlock();
+       return 1;
+end_channel_error:
+       if (channel) {
+               /*
+                * Free channel here since no one has a reference to it. We don't
+                * free after that because a stream can store this pointer.
+                */
+               destroy_channel(channel);
+       }
+       /* We have to send a status channel message indicating an error. */
+       ret = consumer_send_status_channel(sock, NULL);
+       if (ret < 0) {
+               /* Stop everything if session daemon can not be notified. */
+               goto error_fatal;
+       }
+       rcu_read_unlock();
+       return 1;
+error_fatal:
+       rcu_read_unlock();
+       /* This will issue a consumer stop. */
+       return -1;
 }
 
-int lttng_ustconsumer_allocate_channel(struct lttng_consumer_channel *chan)
+/*
+ * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
+ * compiled out, we isolate it in this library.
+ */
+int lttng_ustctl_get_mmap_read_offset(struct lttng_consumer_stream *stream,
+               unsigned long *off)
 {
-       struct lttng_ust_object_data obj;
-
-       obj.handle = -1;
-       obj.shm_fd = chan->shm_fd;
-       obj.wait_fd = chan->wait_fd;
-       obj.memory_map_size = chan->mmap_len;
-       chan->handle = ustctl_map_channel(&obj);
-       if (!chan->handle) {
-               return -ENOMEM;
-       }
-       chan->wait_fd_is_copy = 1;
-       chan->shm_fd = -1;
+       assert(stream);
+       assert(stream->ustream);
 
-       return 0;
+       return ustctl_get_mmap_read_offset(stream->ustream, off);
 }
 
-void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
+/*
+ * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
+ * compiled out, we isolate it in this library.
+ */
+void *lttng_ustctl_get_mmap_base(struct lttng_consumer_stream *stream)
 {
-       ustctl_flush_buffer(stream->chan->handle, stream->buf, 0);
-       stream->hangup_flush_done = 1;
+       assert(stream);
+       assert(stream->ustream);
+
+       return ustctl_get_mmap_base(stream->ustream);
 }
 
-void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
+/*
+ * Take a snapshot for a specific fd
+ *
+ * Returns 0 on success, < 0 on error
+ */
+int lttng_ustconsumer_take_snapshot(struct lttng_consumer_stream *stream)
 {
-       ustctl_unmap_channel(chan->handle);
+       assert(stream);
+       assert(stream->ustream);
+
+       return ustctl_snapshot(stream->ustream);
 }
 
-int lttng_ustconsumer_add_stream(struct lttng_consumer_stream *stream)
+/*
+ * Get the produced position
+ *
+ * Returns 0 on success, < 0 on error
+ */
+int lttng_ustconsumer_get_produced_snapshot(
+               struct lttng_consumer_stream *stream, unsigned long *pos)
 {
-       struct lttng_ust_object_data obj;
-       int ret;
-
-       obj.handle = -1;
-       obj.shm_fd = stream->shm_fd;
-       obj.wait_fd = stream->wait_fd;
-       obj.memory_map_size = stream->mmap_len;
-       ret = ustctl_add_stream(stream->chan->handle, &obj);
-       if (ret) {
-               ERR("UST ctl add_stream failed with ret %d", ret);
-               goto error;
-       }
+       assert(stream);
+       assert(stream->ustream);
+       assert(pos);
 
-       stream->buf = ustctl_open_stream_read(stream->chan->handle, stream->cpu);
-       if (!stream->buf) {
-               ERR("UST ctl open_stream_read failed");
-               ret = -EBUSY;
-               goto error;
-       }
+       return ustctl_snapshot_get_produced(stream->ustream, pos);
+}
 
-       /* ustctl_open_stream_read has closed the shm fd. */
-       stream->wait_fd_is_copy = 1;
-       stream->shm_fd = -1;
+/*
+ * Called when the stream signal the consumer that it has hang up.
+ */
+void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
+{
+       assert(stream);
+       assert(stream->ustream);
 
-       stream->mmap_base = ustctl_get_mmap_base(stream->chan->handle, stream->buf);
-       if (!stream->mmap_base) {
-               ERR("UST ctl get_mmap_base failed");
-               ret = -EINVAL;
-               goto mmap_error;
-       }
+       ustctl_flush_buffer(stream->ustream, 0);
+       stream->hangup_flush_done = 1;
+}
 
-       return 0;
+void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
+{
+       assert(chan);
+       assert(chan->uchan);
 
-mmap_error:
-       ustctl_close_stream_read(stream->chan->handle, stream->buf);
-error:
-       return ret;
+       ustctl_destroy_channel(chan->uchan);
 }
 
 void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
 {
-       ustctl_close_stream_read(stream->chan->handle, stream->buf);
-}
+       assert(stream);
+       assert(stream->ustream);
 
+       ustctl_destroy_stream(stream->ustream);
+}
 
 int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
                struct lttng_consumer_local_data *ctx)
@@ -494,12 +886,18 @@ int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
        unsigned long len, subbuf_size, padding;
        int err;
        long ret = 0;
-       struct lttng_ust_shm_handle *handle;
-       struct lttng_ust_lib_ring_buffer *buf;
        char dummy;
+       struct ustctl_consumer_stream *ustream;
+
+       assert(stream);
+       assert(stream->ustream);
+       assert(ctx);
 
-       DBG("In read_subbuffer (wait_fd: %d, stream key: %d)",
-               stream->wait_fd, stream->key);
+       DBG2("In UST read_subbuffer (wait_fd: %d, name: %s)", stream->wait_fd,
+                       stream->name);
+
+       /* Ease our life for what's next. */
+       ustream = stream->ustream;
 
        /* We can consume the 1 byte written into the wait_fd by UST */
        if (!stream->hangup_flush_done) {
@@ -514,10 +912,8 @@ int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
                }
        }
 
-       buf = stream->buf;
-       handle = stream->chan->handle;
        /* Get the next subbuffer */
-       err = ustctl_get_next_subbuf(handle, buf);
+       err = ustctl_get_next_subbuf(ustream);
        if (err != 0) {
                ret = err;      /* ustctl_get_next_subbuf returns negative, caller expect positive. */
                /*
@@ -527,16 +923,16 @@ int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
                 * would issue wakeups.
                 */
                DBG("Reserving sub buffer failed (everything is normal, "
-                               "it is due to concurrency)");
+                               "it is due to concurrency) [ret: %d]", err);
                goto end;
        }
-       assert(stream->output == LTTNG_EVENT_MMAP);
+       assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
        /* Get the full padded subbuffer size */
-       err = ustctl_get_padded_subbuf_size(handle, buf, &len);
+       err = ustctl_get_padded_subbuf_size(ustream, &len);
        assert(err == 0);
 
        /* Get subbuffer data size (without padding) */
-       err = ustctl_get_subbuf_size(handle, buf, &subbuf_size);
+       err = ustctl_get_subbuf_size(ustream, &subbuf_size);
        assert(err == 0);
 
        /* Make sure we don't get a subbuffer size bigger than the padded */
@@ -563,37 +959,41 @@ int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
                                "(ret: %zd != len: %lu != subbuf_size: %lu)",
                                ret, len, subbuf_size);
        }
-       err = ustctl_put_next_subbuf(handle, buf);
+       err = ustctl_put_next_subbuf(ustream);
        assert(err == 0);
 end:
        return ret;
 }
 
+/*
+ * Called when a stream is created.
+ */
 int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
 {
        int ret;
+       char full_path[PATH_MAX];
 
        /* Opening the tracefile in write mode */
-       if (stream->path_name != NULL && stream->net_seq_idx == -1) {
-               ret = run_as_open(stream->path_name,
-                               O_WRONLY|O_CREAT|O_TRUNC,
-                               S_IRWXU|S_IRWXG|S_IRWXO,
-                               stream->uid, stream->gid);
-               if (ret < 0) {
-                       ERR("Opening %s", stream->path_name);
-                       PERROR("open");
-                       goto error;
-               }
-               stream->out_fd = ret;
+       if (stream->net_seq_idx != -1) {
+               goto end;
        }
 
-       ret = lttng_ustconsumer_add_stream(stream);
-       if (ret) {
-               consumer_del_stream(stream, NULL);
-               ret = -1;
+       ret = snprintf(full_path, sizeof(full_path), "%s/%s",
+                       stream->chan->pathname, stream->name);
+       if (ret < 0) {
+               PERROR("snprintf on_recv_stream");
+               goto error;
+       }
+
+       ret = run_as_open(full_path, O_WRONLY | O_CREAT | O_TRUNC,
+                       S_IRWXU | S_IRWXG | S_IRWXO, stream->uid, stream->gid);
+       if (ret < 0) {
+               PERROR("open stream path %s", full_path);
                goto error;
        }
+       stream->out_fd = ret;
 
+end:
        /* we return 0 to let the library handle the FD internally */
        return 0;
 
@@ -614,13 +1014,14 @@ int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream)
        int ret;
 
        assert(stream);
+       assert(stream->ustream);
 
        DBG("UST consumer checking data pending");
 
-       ret = ustctl_get_next_subbuf(stream->chan->handle, stream->buf);
+       ret = ustctl_get_next_subbuf(stream->ustream);
        if (ret == 0) {
                /* There is still data so let's put back this subbuffer. */
-               ret = ustctl_put_subbuf(stream->chan->handle, stream->buf);
+               ret = ustctl_put_subbuf(stream->ustream);
                assert(ret == 0);
                ret = 1;  /* Data is pending */
                goto end;
index 812c98df4e7f4ad42516a4050a9aa88f737de608..009fa5e5901bcfaa6cd3d3b3d59ba354cb05f06a 100644 (file)
 
 #ifdef HAVE_LIBLTTNG_UST_CTL
 
-/*
- * Take a snapshot for a specific fd
- *
- * Returns 0 on success, < 0 on error
- */
-int lttng_ustconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
-        struct lttng_consumer_stream *stream);
+int lttng_ustconsumer_take_snapshot(struct lttng_consumer_stream *stream);
 
-/*
- * Get the produced position
- *
- * Returns 0 on success, < 0 on error
- */
 int lttng_ustconsumer_get_produced_snapshot(
-        struct lttng_consumer_local_data *ctx,
-        struct lttng_consumer_stream *stream,
-        unsigned long *pos);
+               struct lttng_consumer_stream *stream, unsigned long *pos);
 
 int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                int sock, struct pollfd *consumer_sockpoll);
@@ -58,9 +45,9 @@ int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream);
 
 void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream);
 
-extern int lttng_ustctl_get_mmap_read_offset(
-               struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf, unsigned long *off);
+int lttng_ustctl_get_mmap_read_offset(struct lttng_consumer_stream *stream,
+               unsigned long *off);
+void *lttng_ustctl_get_mmap_base(struct lttng_consumer_stream *stream);
 int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream);
 
 #else /* HAVE_LIBLTTNG_UST_CTL */
@@ -84,17 +71,14 @@ ssize_t lttng_ustconsumer_on_read_subbuffer_splice(
 }
 
 static inline
-int lttng_ustconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
-        struct lttng_consumer_stream *stream)
+int lttng_ustconsumer_take_snapshot(struct lttng_consumer_stream *stream)
 {
        return -ENOSYS;
 }
 
 static inline
 int lttng_ustconsumer_get_produced_snapshot(
-        struct lttng_consumer_local_data *ctx,
-        struct lttng_consumer_stream *stream,
-        unsigned long *pos)
+               struct lttng_consumer_stream *stream, unsigned long *pos)
 {
        return -ENOSYS;
 }
@@ -147,8 +131,8 @@ void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
 }
 
 static inline
-int lttng_ustctl_get_mmap_read_offset(struct lttng_ust_shm_handle *handle,
-               struct lttng_ust_lib_ring_buffer *buf, unsigned long *off)
+int lttng_ustctl_get_mmap_read_offset(struct lttng_consumer_stream *stream,
+               unsigned long *off)
 {
        return -ENOSYS;
 }
@@ -157,6 +141,11 @@ int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream)
 {
        return -ENOSYS;
 }
+static inline
+void *lttng_ustctl_get_mmap_base(struct lttng_consumer_stream *stream)
+{
+       return NULL;
+}
 #endif /* HAVE_LIBLTTNG_UST_CTL */
 
 #endif /* _LTTNG_USTCONSUMER_H */
index 6dffd783eee8d6f9e695b3556bfc56627e61292f..dbf5228ce524a630d50e9b7b0145c3243db977bc 100755 (executable)
@@ -2,7 +2,7 @@
 
 DIR=$(dirname $0)
 
-tests=( $DIR/unit_tests $DIR/uri_switch $DIR/run-kernel $DIR/run-ust )
+tests=( $DIR/unit_tests $DIR/run-kernel $DIR/run-ust )
 exit_code=0
 
 function start_tests ()
This page took 0.174238 seconds and 4 git commands to generate.