Add ustctl_flush_buffer to the consumer API

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

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index ef88a97a405ddd9429d1d3c09abb80bb75960f37..775cb1766f62e4a356cf49ca739351c4cedc7dac 100644 (file)
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -184,6 +184,7 @@ static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
                goto error;
        }
 
+       consumer_stream_update_channel_attributes(stream, channel);
        stream->chan = channel;
 
 error:
@@ -206,20 +207,10 @@ static int send_stream_to_thread(struct lttng_consumer_stream *stream,
 
        /* Get the right pipe where the stream will be sent. */
        if (stream->metadata_flag) {
-               ret = consumer_add_metadata_stream(stream);
-               if (ret) {
-                       ERR("Consumer add metadata stream %" PRIu64 " failed.",
-                                       stream->key);
-                       goto error;
-               }
+               consumer_add_metadata_stream(stream);
                stream_pipe = ctx->consumer_metadata_pipe;
        } else {
-               ret = consumer_add_data_stream(stream);
-               if (ret) {
-                       ERR("Consumer add stream %" PRIu64 " failed.",
-                                       stream->key);
-                       goto error;
-               }
+               consumer_add_data_stream(stream);
                stream_pipe = ctx->consumer_data_pipe;
        }
 
@@ -830,6 +821,7 @@ static int close_metadata(uint64_t chan_key)
 {
        int ret = 0;
        struct lttng_consumer_channel *channel;
+       unsigned int channel_monitor;
 
        DBG("UST consumer close metadata key %" PRIu64, chan_key);
 
@@ -848,13 +840,48 @@ static int close_metadata(uint64_t chan_key)
 
        pthread_mutex_lock(&consumer_data.lock);
        pthread_mutex_lock(&channel->lock);
-
+       channel_monitor = channel->monitor;
        if (cds_lfht_is_node_deleted(&channel->node.node)) {
                goto error_unlock;
        }
 
        lttng_ustconsumer_close_metadata(channel);
+       pthread_mutex_unlock(&channel->lock);
+       pthread_mutex_unlock(&consumer_data.lock);
 
+       /*
+        * The ownership of a metadata channel depends on the type of
+        * session to which it belongs. In effect, the monitor flag is checked
+        * to determine if this metadata channel is in "snapshot" mode or not.
+        *
+        * In the non-snapshot case, the metadata channel is created along with
+        * a single stream which will remain present until the metadata channel
+        * is destroyed (on the destruction of its session). In this case, the
+        * metadata stream in "monitored" by the metadata poll thread and holds
+        * the ownership of its channel.
+        *
+        * Closing the metadata will cause the metadata stream's "metadata poll
+        * pipe" to be closed. Closing this pipe will wake-up the metadata poll
+        * thread which will teardown the metadata stream which, in return,
+        * deletes the metadata channel.
+        *
+        * In the snapshot case, the metadata stream is created and destroyed
+        * on every snapshot record. Since the channel doesn't have an owner
+        * other than the session daemon, it is safe to destroy it immediately
+        * on reception of the CLOSE_METADATA command.
+        */
+       if (!channel_monitor) {
+               /*
+                * The channel and consumer_data locks must be
+                * released before this call since consumer_del_channel
+                * re-acquires the channel and consumer_data locks to teardown
+                * the channel and queue its reclamation by the "call_rcu"
+                * worker thread.
+                */
+               consumer_del_channel(channel);
+       }
+
+       return ret;
 error_unlock:
        pthread_mutex_unlock(&channel->lock);
        pthread_mutex_unlock(&consumer_data.lock);
@@ -1891,6 +1918,52 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                }
                goto end_msg_sessiond;
        }
+       case LTTNG_CONSUMER_ROTATE_RENAME:
+       {
+               DBG("Consumer rename session %" PRIu64 " after rotation",
+                               msg.u.rotate_rename.session_id);
+               ret = lttng_consumer_rotate_rename(msg.u.rotate_rename.old_path,
+                               msg.u.rotate_rename.new_path,
+                               msg.u.rotate_rename.uid,
+                               msg.u.rotate_rename.gid,
+                               msg.u.rotate_rename.relayd_id);
+               if (ret < 0) {
+                       ERR("Rotate rename failed");
+                       ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+               }
+
+               health_code_update();
+
+               ret = consumer_send_status_msg(sock, ret_code);
+               if (ret < 0) {
+                       /* Somehow, the session daemon is not responding anymore. */
+                       goto end_nosignal;
+               }
+               break;
+       }
+       case LTTNG_CONSUMER_MKDIR:
+       {
+               DBG("Consumer mkdir %s in session %" PRIu64,
+                               msg.u.mkdir.path,
+                               msg.u.mkdir.session_id);
+               ret = lttng_consumer_mkdir(msg.u.mkdir.path,
+                               msg.u.mkdir.uid,
+                               msg.u.mkdir.gid,
+                               msg.u.mkdir.relayd_id);
+               if (ret < 0) {
+                       ERR("consumer mkdir failed");
+                       ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+               }
+
+               health_code_update();
+
+               ret = consumer_send_status_msg(sock, ret_code);
+               if (ret < 0) {
+                       /* Somehow, the session daemon is not responding anymore. */
+                       goto end_nosignal;
+               }
+               break;
+       }
        default:
                break;
        }
@@ -1971,6 +2044,15 @@ void *lttng_ustctl_get_mmap_base(struct lttng_consumer_stream *stream)
        return ustctl_get_mmap_base(stream->ustream);
 }
 
+void lttng_ustctl_flush_buffer(struct lttng_consumer_stream *stream,
+               int producer_active)
+{
+       assert(stream);
+       assert(stream->ustream);
+
+       ustctl_flush_buffer(stream->ustream, producer_active);
+}
+
 /*
  * Take a snapshot for a specific stream.
  *