Relayd data available command support

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

diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c
index 13cbe2149de6852ed08e6532f6025ff3e382a103..4e000fbd167e294c6b9139b6b7bdd2cf0098d6f1 100644 (file)
--- a/src/common/kernel-consumer/kernel-consumer.c
+++ b/src/common/kernel-consumer/kernel-consumer.c
@@ -138,7 +138,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
        }
        case LTTNG_CONSUMER_ADD_STREAM:
        {
-               int fd;
+               int fd, stream_pipe;
                struct consumer_relayd_sock_pair *relayd = NULL;
                struct lttng_consumer_stream *new_stream;
                int alloc_ret = 0;
@@ -168,6 +168,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                                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) {
@@ -224,27 +225,26 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                        }
                }
 
-               /* Send stream to the metadata thread */
+               /* Get the right pipe where the stream will be sent. */
                if (new_stream->metadata_flag) {
-                       do {
-                               ret = write(ctx->consumer_metadata_pipe[1], &new_stream,
-                                               sizeof(new_stream));
-                       } while (ret < 0 && errno == EINTR);
-                       if (ret < 0) {
-                               PERROR("write metadata pipe");
-                               consumer_del_stream(new_stream, NULL);
-                       }
+                       stream_pipe = ctx->consumer_metadata_pipe[1];
                } else {
-                       ret = consumer_add_stream(new_stream);
-                       if (ret) {
-                               ERR("Consumer add stream %d failed. Continuing",
-                                               new_stream->key);
-                               consumer_del_stream(new_stream, NULL);
-                               goto end_nosignal;
-                       }
+                       stream_pipe = ctx->consumer_data_pipe[1];
                }
 
-               DBG("Kernel consumer_add_stream (%d)", fd);
+               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("Kernel consumer ADD_STREAM %s (fd: %d) with relayd id %" PRIu64,
+                               msg.u.stream.path_name, fd, new_stream->relayd_stream_id);
                break;
        }
        case LTTNG_CONSUMER_UPDATE_STREAM:
@@ -280,24 +280,26 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 
                goto end_nosignal;
        }
+       case LTTNG_CONSUMER_DATA_AVAILABLE:
+       {
+               int32_t ret;
+               uint64_t id = msg.u.data_available.session_id;
+
+               DBG("Kernel consumer data available command for id %" PRIu64, id);
+
+               ret = consumer_data_available(id);
+
+               /* Send back returned value to session daemon */
+               ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret));
+               if (ret < 0) {
+                       PERROR("send data available ret code");
+               }
+               break;
+       }
        default:
                goto end_nosignal;
        }
 
-       /*
-        * Wake-up the other end by writing a null byte in the pipe (non-blocking).
-        * Important note: Because writing into the pipe is non-blocking (and
-        * therefore we allow dropping wakeup data, as long as there is wakeup data
-        * present in the pipe buffer to wake up the other end), the other end
-        * should perform the following sequence for waiting:
-        *
-        * 1) empty the pipe (reads).
-        * 2) perform update operation.
-        * 3) wait on the pipe (poll).
-        */
-       do {
-               ret = write(ctx->consumer_poll_pipe[1], "", 1);
-       } while (ret < 0 && errno == EINTR);
 end_nosignal:
        rcu_read_unlock();
 
@@ -481,3 +483,42 @@ error:
        return ret;
 }
 
+/*
+ * Check if data is still being extracted from the buffers for a specific
+ * stream. Consumer data lock MUST be acquired before calling this function.
+ *
+ * Return 0 if the traced data are still getting read else 1 meaning that the
+ * data is available for trace viewer reading.
+ */
+int lttng_kconsumer_data_available(struct lttng_consumer_stream *stream)
+{
+       int ret;
+
+       assert(stream);
+
+       /*
+        * Try to lock the stream mutex. On failure, we know that the stream is
+        * being used else where hence there is data still being extracted.
+        */
+       ret = pthread_mutex_trylock(&stream->lock);
+       if (ret == EBUSY) {
+               goto data_not_available;
+       }
+       /* The stream is now locked so we can do our ustctl calls */
+
+       ret = kernctl_get_next_subbuf(stream->wait_fd);
+       if (ret == 0) {
+               /* There is still data so let's put back this subbuffer. */
+               ret = kernctl_put_subbuf(stream->wait_fd);
+               assert(ret == 0);
+               pthread_mutex_unlock(&stream->lock);
+               goto data_not_available;
+       }
+
+       /* Data is available to be read for this stream. */
+       pthread_mutex_unlock(&stream->lock);
+       return 1;
+
+data_not_available:
+       return 0;
+}