Fix: code refactoring of viewer streams in relayd
authorDavid Goulet <dgoulet@efficios.com>
Mon, 20 Jan 2014 17:24:26 +0000 (12:24 -0500)
committerDavid Goulet <dgoulet@efficios.com>
Fri, 7 Feb 2014 20:27:33 +0000 (15:27 -0500)
This also renames the lttng-viewer.h to lttng-viewer-abi.h which
represents the live reading ABI.

Acked-by: Julien Desfossez <julien.desfossez@efficios.com>
Signed-off-by: David Goulet <dgoulet@efficios.com>
13 files changed:
src/bin/lttng-relayd/Makefile.am
src/bin/lttng-relayd/live.c
src/bin/lttng-relayd/lttng-relayd.h
src/bin/lttng-relayd/lttng-viewer-abi.h [new file with mode: 0644]
src/bin/lttng-relayd/lttng-viewer.h [deleted file]
src/bin/lttng-relayd/main.c
src/bin/lttng-relayd/session.c [new file with mode: 0644]
src/bin/lttng-relayd/session.h [new file with mode: 0644]
src/bin/lttng-relayd/viewer-stream.c [new file with mode: 0644]
src/bin/lttng-relayd/viewer-stream.h [new file with mode: 0644]
src/common/index/index.c
src/common/index/index.h
tests/regression/tools/live/live_test.c

index f940e3f90085dd218d32e6e3a34175a6e4a27271..65e3d675c012b352dca06ea1f04c596cb6879f47 100644 (file)
@@ -13,7 +13,9 @@ lttng_relayd_SOURCES = main.c lttng-relayd.h utils.h utils.c cmd.h \
                        cmd-2-2.c cmd-2-2.h \
                        cmd-2-4.c cmd-2-4.h \
                        health-relayd.c health-relayd.h \
-                       lttng-viewer.h testpoint.h
+                       lttng-viewer-abi.h testpoint.h \
+                       viewer-stream.h viewer-stream.c \
+                       session.c session.h
 
 # link on liblttngctl for check if relayd is already alive.
 lttng_relayd_LDADD = -lrt -lurcu-common -lurcu \
index 69e2eb3b286c2f54cad19fefbdbe1e5f2bb93492..d558b71487813aadf6206427c5a2f658589bbcb7 100644 (file)
@@ -45,6 +45,7 @@
 #include <common/compat/socket.h>
 #include <common/defaults.h>
 #include <common/futex.h>
+#include <common/index/index.h>
 #include <common/sessiond-comm/sessiond-comm.h>
 #include <common/sessiond-comm/inet.h>
 #include <common/sessiond-comm/relayd.h>
 #include "cmd.h"
 #include "live.h"
 #include "lttng-relayd.h"
-#include "lttng-viewer.h"
 #include "utils.h"
 #include "health-relayd.h"
 #include "testpoint.h"
+#include "viewer-stream.h"
 
 static struct lttng_uri *live_uri;
 
@@ -95,6 +96,215 @@ void cleanup(void)
        free(live_uri);
 }
 
+/*
+ * Receive a request buffer using a given socket, destination allocated buffer
+ * of length size.
+ *
+ * Return the size of the received message or else a negative value on error
+ * with errno being set by recvmsg() syscall.
+ */
+static
+ssize_t recv_request(struct lttcomm_sock *sock, void *buf, size_t size)
+{
+       ssize_t ret;
+
+       assert(sock);
+       assert(buf);
+
+       ret = sock->ops->recvmsg(sock, buf, size, 0);
+       if (ret < 0 || ret != size) {
+               if (ret == 0) {
+                       /* Orderly shutdown. Not necessary to print an error. */
+                       DBG("Socket %d did an orderly shutdown", sock->fd);
+               } else {
+                       ERR("Relay failed to receive request.");
+               }
+               ret = -1;
+       }
+
+       return ret;
+}
+
+/*
+ * Send a response buffer using a given socket, source allocated buffer of
+ * length size.
+ *
+ * Return the size of the sent message or else a negative value on error with
+ * errno being set by sendmsg() syscall.
+ */
+static
+ssize_t send_response(struct lttcomm_sock *sock, void *buf, size_t size)
+{
+       ssize_t ret;
+
+       assert(sock);
+       assert(buf);
+
+       ret = sock->ops->sendmsg(sock, buf, size, 0);
+       if (ret < 0) {
+               ERR("Relayd failed to send response.");
+       }
+
+       return ret;
+}
+
+/*
+ * Atomically check if new streams got added in the session since the last
+ * check and reset the flag to 0.
+ *
+ * Returns 1 if new streams got added, 0 if nothing changed, a negative value
+ * on error.
+ */
+static
+int check_new_streams(uint64_t session_id, struct lttng_ht *sessions_ht)
+{
+       int ret;
+       unsigned long current_val;
+       struct relay_session *session;
+
+       assert(sessions_ht);
+
+       session = session_find_by_id(sessions_ht, session_id);
+       if (!session) {
+               DBG("Relay session %" PRIu64 " not found", session_id);
+               ret = -1;
+               goto error;
+       }
+
+       current_val = uatomic_cmpxchg(&session->new_streams, 1, 0);
+       ret = current_val;
+
+error:
+       return ret;
+}
+
+/*
+ * Send viewer streams to the given socket. The ignore_sent_flag indicates if
+ * this function should ignore the sent flag or not.
+ *
+ * Return 0 on success or else a negative value.
+ */
+static
+ssize_t send_viewer_streams(struct lttcomm_sock *sock,
+               struct relay_session *session, unsigned int ignore_sent_flag)
+{
+       ssize_t ret;
+       struct lttng_viewer_stream send_stream;
+       struct lttng_ht_iter iter;
+       struct relay_viewer_stream *vstream;
+
+       assert(session);
+
+       rcu_read_lock();
+
+       cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, vstream,
+                       stream_n.node) {
+               health_code_update();
+
+               /* Ignore if not the same session. */
+               if (vstream->session_id != session->id ||
+                               (!ignore_sent_flag && vstream->sent_flag)) {
+                       continue;
+               }
+
+               send_stream.id = htobe64(vstream->stream_handle);
+               send_stream.ctf_trace_id = htobe64(vstream->ctf_trace->id);
+               send_stream.metadata_flag = htobe32(vstream->metadata_flag);
+               strncpy(send_stream.path_name, vstream->path_name,
+                               sizeof(send_stream.path_name));
+               strncpy(send_stream.channel_name, vstream->channel_name,
+                               sizeof(send_stream.channel_name));
+
+               DBG("Sending stream %" PRIu64 " to viewer", vstream->stream_handle);
+               ret = send_response(sock, &send_stream, sizeof(send_stream));
+               if (ret < 0) {
+                       goto end_unlock;
+               }
+               vstream->sent_flag = 1;
+       }
+
+       ret = 0;
+
+end_unlock:
+       rcu_read_unlock();
+       return ret;
+}
+
+/*
+ * Create every viewer stream possible for the given session with the seek
+ * type. Three counters *can* be return which are in order the total amount of
+ * viewer stream of the session, the number of unsent stream and the number of
+ * stream created. Those counters can be NULL and thus will be ignored.
+ *
+ * Return 0 on success or else a negative value.
+ */
+static
+int make_viewer_streams(struct relay_session *session,
+               enum lttng_viewer_seek seek_t, uint32_t *nb_total, uint32_t *nb_unsent,
+               uint32_t *nb_created)
+{
+       int ret;
+       struct relay_stream *stream;
+       struct lttng_ht_iter iter;
+
+       assert(session);
+
+       /*
+        * This is to make sure we create viewer streams for a full received
+        * channel. For instance, if we have 8 streams for a channel that are
+        * concurrently being flagged ready, we can end up creating just a subset
+        * of the 8 streams (the ones that are flagged). This lock avoids this
+        * limbo state.
+        */
+       pthread_mutex_lock(&session->viewer_ready_lock);
+
+       /*
+        * Create viewer streams for relay streams that are ready to be used for a
+        * the given session id only.
+        */
+       cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
+                       stream_n.node) {
+               struct relay_viewer_stream *vstream;
+
+               health_code_update();
+
+               if (stream->session->id != session->id ||
+                               !stream->ctf_trace || !stream->viewer_ready) {
+                       /*
+                        * Ignore stream from a different session. Don't create streams
+                        * with no ctf_trace or not ready for the viewer.
+                        */
+                       continue;
+               }
+
+               vstream = viewer_stream_find_by_id(stream->stream_handle);
+               if (!vstream) {
+                       vstream = viewer_stream_create(stream, seek_t);
+                       if (!vstream) {
+                               ret = -1;
+                               goto error_unlock;
+                       }
+                       if (nb_created) {
+                               /* Update number of created stream counter. */
+                               (*nb_created)++;
+                       }
+               } else if (!vstream->sent_flag && nb_unsent) {
+                       /* Update number of unsent stream counter. */
+                       (*nb_unsent)++;
+               }
+               /* Update number of total stream counter. */
+               if (nb_total) {
+                       (*nb_total)++;
+               }
+       }
+
+       ret = 0;
+
+error_unlock:
+       pthread_mutex_unlock(&session->viewer_ready_lock);
+       return ret;
+}
+
 /*
  * Write to writable pipe used to notify a thread.
  */
@@ -461,16 +671,10 @@ int viewer_connect(struct relay_command *cmd)
 
        health_code_update();
 
-       /* Get version from the other side. */
-       ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
-       if (ret < 0 || ret != sizeof(msg)) {
-               if (ret == 0) {
-                       /* Orderly shutdown. Not necessary to print an error. */
-                       DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
-               } else {
-                       ERR("Relay failed to receive the version values.");
-               }
-               ret = -1;
+       DBG("Viewer is establishing a connection to the relayd.");
+
+       ret = recv_request(cmd->sock, &msg, sizeof(msg));
+       if (ret < 0) {
                goto end;
        }
 
@@ -481,8 +685,8 @@ int viewer_connect(struct relay_command *cmd)
 
        /* Major versions must be the same */
        if (reply.major != be32toh(msg.major)) {
-               DBG("Incompatible major versions (%u vs %u)", reply.major,
-                               be32toh(msg.major));
+               DBG("Incompatible major versions ([relayd] %u vs [client] %u)",
+                               reply.major, be32toh(msg.major));
                ret = -1;
                goto end;
        }
@@ -513,10 +717,10 @@ int viewer_connect(struct relay_command *cmd)
 
        health_code_update();
 
-       ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
-                       sizeof(struct lttng_viewer_connect), 0);
+
+       ret = send_response(cmd->sock, &reply, sizeof(reply));
        if (ret < 0) {
-               ERR("Relay sending version");
+               goto end;
        }
 
        health_code_update();
@@ -541,43 +745,29 @@ int viewer_list_sessions(struct relay_command *cmd,
        struct lttng_viewer_list_sessions session_list;
        unsigned long count;
        long approx_before, approx_after;
-       struct lttng_ht_node_ulong *node;
        struct lttng_ht_iter iter;
        struct lttng_viewer_session send_session;
        struct relay_session *session;
 
        DBG("List sessions received");
 
-       if (cmd->version_check_done == 0) {
-               ERR("Trying to list sessions before version check");
-               ret = -1;
-               goto end_no_session;
-       }
-
        rcu_read_lock();
        cds_lfht_count_nodes(sessions_ht->ht, &approx_before, &count, &approx_after);
        session_list.sessions_count = htobe32(count);
 
        health_code_update();
 
-       ret = cmd->sock->ops->sendmsg(cmd->sock, &session_list,
-                       sizeof(session_list), 0);
+       ret = send_response(cmd->sock, &session_list, sizeof(session_list));
        if (ret < 0) {
-               ERR("Relay sending sessions list");
                goto end_unlock;
        }
 
        health_code_update();
 
-       cds_lfht_for_each_entry(sessions_ht->ht, &iter.iter, node, node) {
+       cds_lfht_for_each_entry(sessions_ht->ht, &iter.iter, session,
+                       session_n.node) {
                health_code_update();
 
-               node = lttng_ht_iter_get_node_ulong(&iter);
-               if (!node) {
-                       goto end_unlock;
-               }
-               session = caa_container_of(node, struct relay_session, session_n);
-
                strncpy(send_session.session_name, session->session_name,
                                sizeof(send_session.session_name));
                strncpy(send_session.hostname, session->hostname,
@@ -589,10 +779,8 @@ int viewer_list_sessions(struct relay_command *cmd,
 
                health_code_update();
 
-               ret = cmd->sock->ops->sendmsg(cmd->sock, &send_session,
-                               sizeof(send_session), 0);
+               ret = send_response(cmd->sock, &send_session, sizeof(send_session));
                if (ret < 0) {
-                       ERR("Relay sending session info");
                        goto end_unlock;
                }
        }
@@ -606,249 +794,6 @@ end_unlock:
        rcu_read_unlock();
 
 end:
-end_no_session:
-       return ret;
-}
-
-/*
- * Open index file using a given viewer stream.
- *
- * Return 0 on success or else a negative value.
- */
-static int open_index(struct relay_viewer_stream *stream)
-{
-       int ret;
-       char fullpath[PATH_MAX];
-       struct ctf_packet_index_file_hdr hdr;
-
-       if (stream->tracefile_count > 0) {
-               ret = snprintf(fullpath, sizeof(fullpath), "%s/" DEFAULT_INDEX_DIR "/%s_%"
-                               PRIu64 DEFAULT_INDEX_FILE_SUFFIX, stream->path_name,
-                               stream->channel_name, stream->tracefile_count_current);
-       } else {
-               ret = snprintf(fullpath, sizeof(fullpath), "%s/" DEFAULT_INDEX_DIR "/%s"
-                               DEFAULT_INDEX_FILE_SUFFIX, stream->path_name,
-                               stream->channel_name);
-       }
-       if (ret < 0) {
-               PERROR("snprintf index path");
-               goto error;
-       }
-
-       DBG("Opening index file %s in read only", fullpath);
-       ret = open(fullpath, O_RDONLY);
-       if (ret < 0) {
-               if (errno == ENOENT) {
-                       ret = -ENOENT;
-                       goto error;
-               } else {
-                       PERROR("opening index in read-only");
-               }
-               goto error;
-       }
-       stream->index_read_fd = ret;
-       DBG("Opening index file %s in read only, (fd: %d)", fullpath, ret);
-
-       ret = lttng_read(stream->index_read_fd, &hdr, sizeof(hdr));
-       if (ret < sizeof(hdr)) {
-               PERROR("Reading index header");
-               goto error;
-       }
-       if (be32toh(hdr.magic) != CTF_INDEX_MAGIC) {
-               ERR("Invalid header magic");
-               ret = -1;
-               goto error;
-       }
-       if (be32toh(hdr.index_major) != CTF_INDEX_MAJOR ||
-                       be32toh(hdr.index_minor) != CTF_INDEX_MINOR) {
-               ERR("Invalid header version");
-               ret = -1;
-               goto error;
-       }
-       ret = 0;
-
-error:
-       return ret;
-}
-
-/*
- * Allocate and init a new viewer_stream.
- *
- * Copies the values from the stream passed in parameter and insert the new
- * stream in the viewer_streams_ht.
- *
- * MUST be called with rcu_read_lock held.
- *
- * Returns 0 on success or a negative value on error.
- */
-static
-int init_viewer_stream(struct relay_stream *stream, int seek_last)
-{
-       int ret;
-       struct relay_viewer_stream *viewer_stream;
-
-       assert(stream);
-
-       viewer_stream = zmalloc(sizeof(*viewer_stream));
-       if (!viewer_stream) {
-               PERROR("relay viewer stream zmalloc");
-               ret = -1;
-               goto error;
-       }
-       viewer_stream->session_id = stream->session->id;
-       viewer_stream->stream_handle = stream->stream_handle;
-       viewer_stream->path_name = strndup(stream->path_name,
-                       LTTNG_VIEWER_PATH_MAX);
-       viewer_stream->channel_name = strndup(stream->channel_name,
-                       LTTNG_VIEWER_NAME_MAX);
-       viewer_stream->tracefile_count = stream->tracefile_count;
-       viewer_stream->metadata_flag = stream->metadata_flag;
-       viewer_stream->tracefile_count_last = -1ULL;
-       if (seek_last) {
-               viewer_stream->tracefile_count_current =
-                       stream->tracefile_count_current;
-       } else {
-               viewer_stream->tracefile_count_current =
-                       stream->oldest_tracefile_id;
-       }
-
-       viewer_stream->ctf_trace = stream->ctf_trace;
-       if (viewer_stream->metadata_flag) {
-               viewer_stream->ctf_trace->viewer_metadata_stream =
-                       viewer_stream;
-       }
-       uatomic_inc(&viewer_stream->ctf_trace->refcount);
-
-       lttng_ht_node_init_u64(&viewer_stream->stream_n, stream->stream_handle);
-       lttng_ht_add_unique_u64(viewer_streams_ht, &viewer_stream->stream_n);
-
-       viewer_stream->index_read_fd = -1;
-       viewer_stream->read_fd = -1;
-
-       /*
-        * This is to avoid a race between the initialization of this object and
-        * the close of the given stream. If the stream is unable to find this
-        * viewer stream when closing, this copy will at least take the latest
-        * value.
-        * We also need that for the seek_last.
-        */
-       viewer_stream->total_index_received = stream->total_index_received;
-
-       /*
-        * If we never received an index for the current stream, delay
-        * the opening of the index, otherwise open it right now.
-        */
-       if (viewer_stream->tracefile_count_current ==
-                       stream->tracefile_count_current &&
-                       viewer_stream->total_index_received == 0) {
-               viewer_stream->index_read_fd = -1;
-       } else {
-               ret = open_index(viewer_stream);
-               if (ret < 0) {
-                       goto error;
-               }
-       }
-
-       if (seek_last && viewer_stream->index_read_fd > 0) {
-               ret = lseek(viewer_stream->index_read_fd,
-                               viewer_stream->total_index_received *
-                                       sizeof(struct ctf_packet_index),
-                               SEEK_CUR);
-               if (ret < 0) {
-                       goto error;
-               }
-               viewer_stream->last_sent_index =
-                       viewer_stream->total_index_received;
-       }
-
-       ret = 0;
-
-error:
-       return ret;
-}
-
-/*
- * Rotate a stream to the next tracefile.
- *
- * Returns 0 on success, 1 on EOF, a negative value on error.
- */
-static
-int rotate_viewer_stream(struct relay_viewer_stream *viewer_stream,
-               struct relay_stream *stream)
-{
-       int ret;
-       uint64_t tracefile_id;
-
-       assert(viewer_stream);
-
-       tracefile_id = (viewer_stream->tracefile_count_current + 1) %
-               viewer_stream->tracefile_count;
-       /*
-        * Detect the last tracefile to open.
-        */
-       if (viewer_stream->tracefile_count_last != -1ULL &&
-                       viewer_stream->tracefile_count_last ==
-                       viewer_stream->tracefile_count_current) {
-               ret = 1;
-               goto end;
-       }
-
-       if (stream) {
-               pthread_mutex_lock(&stream->viewer_stream_rotation_lock);
-       }
-       /*
-        * The writer and the reader are not working in the same
-        * tracefile, we can read up to EOF, we don't care about the
-        * total_index_received.
-        */
-       if (!stream || (stream->tracefile_count_current != tracefile_id)) {
-               viewer_stream->close_write_flag = 1;
-       } else {
-               /*
-                * We are opening a file that is still open in write, make
-                * sure we limit our reading to the number of indexes
-                * received.
-                */
-               viewer_stream->close_write_flag = 0;
-               if (stream) {
-                       viewer_stream->total_index_received =
-                               stream->total_index_received;
-               }
-       }
-       viewer_stream->tracefile_count_current = tracefile_id;
-
-       ret = close(viewer_stream->index_read_fd);
-       if (ret < 0) {
-               PERROR("close index file %d",
-                               viewer_stream->index_read_fd);
-       }
-       viewer_stream->index_read_fd = -1;
-       ret = close(viewer_stream->read_fd);
-       if (ret < 0) {
-               PERROR("close tracefile %d",
-                               viewer_stream->read_fd);
-       }
-       viewer_stream->read_fd = -1;
-
-       pthread_mutex_lock(&viewer_stream->overwrite_lock);
-       viewer_stream->abort_flag = 0;
-       pthread_mutex_unlock(&viewer_stream->overwrite_lock);
-
-       viewer_stream->index_read_fd = -1;
-       viewer_stream->read_fd = -1;
-
-       if (stream) {
-               pthread_mutex_unlock(&stream->viewer_stream_rotation_lock);
-       }
-       ret = open_index(viewer_stream);
-       if (ret < 0) {
-               goto error;
-       }
-
-       ret = 0;
-
-end:
-error:
        return ret;
 }
 
@@ -860,14 +805,9 @@ int viewer_get_new_streams(struct relay_command *cmd,
                struct lttng_ht *sessions_ht)
 {
        int ret, send_streams = 0;
-       uint32_t nb_streams = 0;
+       uint32_t nb_created = 0, nb_unsent = 0, nb_streams = 0;
        struct lttng_viewer_new_streams_request request;
        struct lttng_viewer_new_streams_response response;
-       struct lttng_viewer_stream send_stream;
-       struct relay_stream *stream;
-       struct relay_viewer_stream *viewer_stream;
-       struct lttng_ht_node_ulong *node;
-       struct lttng_ht_iter iter;
        struct relay_session *session;
 
        assert(cmd);
@@ -875,40 +815,25 @@ int viewer_get_new_streams(struct relay_command *cmd,
 
        DBG("Get new streams received");
 
-       if (cmd->version_check_done == 0) {
-               ERR("Trying to get streams before version check");
-               ret = -1;
-               goto end_no_session;
-       }
-
        health_code_update();
 
-       ret = cmd->sock->ops->recvmsg(cmd->sock, &request, sizeof(request), 0);
-       if (ret < 0 || ret != sizeof(request)) {
-               if (ret == 0) {
-                       /* Orderly shutdown. Not necessary to print an error. */
-                       DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
-               } else {
-                       ERR("Relay failed to receive the command parameters.");
-               }
-               ret = -1;
+       /* Receive the request from the connected client. */
+       ret = recv_request(cmd->sock, &request, sizeof(request));
+       if (ret < 0) {
                goto error;
        }
 
        health_code_update();
 
        rcu_read_lock();
-       lttng_ht_lookup(sessions_ht,
-                       (void *)((unsigned long) be64toh(request.session_id)), &iter);
-       node = lttng_ht_iter_get_node_ulong(&iter);
-       if (node == NULL) {
+       session = session_find_by_id(sessions_ht, be64toh(request.session_id));
+       if (!session) {
                DBG("Relay session %" PRIu64 " not found",
                                be64toh(request.session_id));
                response.status = htobe32(VIEWER_NEW_STREAMS_ERR);
                goto send_reply;
        }
 
-       session = caa_container_of(node, struct relay_session, session_n);
        if (cmd->session_id == session->id) {
                /* We confirmed the viewer is asking for the same session. */
                send_streams = 1;
@@ -919,48 +844,23 @@ int viewer_get_new_streams(struct relay_command *cmd,
                goto send_reply;
        }
 
-       /*
-        * Fill the viewer_streams_ht to count the number of streams ready to be
-        * sent and avoid concurrency issues on the relay_streams_ht and don't rely
-        * on a total session stream count.
-        */
-       pthread_mutex_lock(&session->viewer_ready_lock);
-       cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
-                       stream_n.node) {
-               struct relay_viewer_stream *vstream;
-
-               health_code_update();
-
-               /*
-                * Don't send stream if no ctf_trace, wrong session or if the stream is
-                * not ready for the viewer.
-                */
-               if (stream->session != cmd->session ||
-                               !stream->ctf_trace || !stream->viewer_ready) {
-                       continue;
-               }
-
-               vstream = live_find_viewer_stream_by_id(stream->stream_handle);
-               if (!vstream) {
-                       ret = init_viewer_stream(stream, 0);
-                       if (ret < 0) {
-                               pthread_mutex_unlock(&session->viewer_ready_lock);
-                               goto end_unlock;
-                       }
-                       nb_streams++;
-               } else if (!vstream->sent_flag) {
-                       nb_streams++;
-               }
+       if (!send_streams) {
+               goto send_reply;
        }
-       pthread_mutex_unlock(&session->viewer_ready_lock);
 
+       ret = make_viewer_streams(session, VIEWER_SEEK_LAST, NULL, &nb_unsent,
+                       &nb_created);
+       if (ret < 0) {
+               goto end_unlock;
+       }
+       /* Only send back the newly created streams with the unsent ones. */
+       nb_streams = nb_created + nb_unsent;
        response.streams_count = htobe32(nb_streams);
 
 send_reply:
        health_code_update();
-       ret = cmd->sock->ops->sendmsg(cmd->sock, &response, sizeof(response), 0);
+       ret = send_response(cmd->sock, &response, sizeof(response));
        if (ret < 0) {
-               ERR("Relay sending viewer attach response");
                goto end_unlock;
        }
        health_code_update();
@@ -974,40 +874,17 @@ send_reply:
                goto end_unlock;
        }
 
-       /* We should only be there if we have a session to attach to. */
-       cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, viewer_stream,
-                       stream_n.node) {
-               health_code_update();
-
-               /* Don't send back if session does not match or already sent. */
-               if (viewer_stream->session_id != cmd->session->id ||
-                               viewer_stream->sent_flag) {
-                       continue;
-               }
-
-               send_stream.id = htobe64(viewer_stream->stream_handle);
-               send_stream.ctf_trace_id = htobe64(viewer_stream->ctf_trace->id);
-               send_stream.metadata_flag = htobe32(viewer_stream->metadata_flag);
-               strncpy(send_stream.path_name, viewer_stream->path_name,
-                               sizeof(send_stream.path_name));
-               strncpy(send_stream.channel_name, viewer_stream->channel_name,
-                               sizeof(send_stream.channel_name));
-
-               ret = cmd->sock->ops->sendmsg(cmd->sock, &send_stream,
-                               sizeof(send_stream), 0);
-               if (ret < 0) {
-                       ERR("Relay sending stream %" PRIu64, viewer_stream->stream_handle);
-                       goto end_unlock;
-               }
-               DBG("Sent stream %" PRIu64 " to viewer", viewer_stream->stream_handle);
-               viewer_stream->sent_flag = 1;
+       /*
+        * Send stream and *DON'T* ignore the sent flag so every viewer streams
+        * that were not sent from that point will be sent to the viewer.
+        */
+       ret = send_viewer_streams(cmd->sock, session, 0);
+       if (ret < 0) {
+               goto end_unlock;
        }
 
-       ret = 0;
-
 end_unlock:
        rcu_read_unlock();
-end_no_session:
 error:
        return ret;
 }
@@ -1019,58 +896,38 @@ static
 int viewer_attach_session(struct relay_command *cmd,
                struct lttng_ht *sessions_ht)
 {
-       int ret, send_streams = 0;
+       int send_streams = 0;
+       ssize_t ret;
        uint32_t nb_streams = 0;
+       enum lttng_viewer_seek seek_type;
        struct lttng_viewer_attach_session_request request;
        struct lttng_viewer_attach_session_response response;
-       struct lttng_viewer_stream send_stream;
-       struct relay_stream *stream;
-       struct relay_viewer_stream *viewer_stream;
-       struct lttng_ht_node_ulong *node;
-       struct lttng_ht_node_u64 *node64;
-       struct lttng_ht_iter iter;
        struct relay_session *session;
-       int seek_last = 0;
 
        assert(cmd);
        assert(sessions_ht);
 
        DBG("Attach session received");
 
-       if (cmd->version_check_done == 0) {
-               ERR("Trying to attach session before version check");
-               ret = -1;
-               goto end_no_session;
-       }
-
        health_code_update();
 
-       ret = cmd->sock->ops->recvmsg(cmd->sock, &request, sizeof(request), 0);
-       if (ret < 0 || ret != sizeof(request)) {
-               if (ret == 0) {
-                       /* Orderly shutdown. Not necessary to print an error. */
-                       DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
-               } else {
-                       ERR("Relay failed to receive the attach parameters.");
-               }
-               ret = -1;
+       /* Receive the request from the connected client. */
+       ret = recv_request(cmd->sock, &request, sizeof(request));
+       if (ret < 0) {
                goto error;
        }
 
        health_code_update();
 
        rcu_read_lock();
-       lttng_ht_lookup(sessions_ht,
-                       (void *)((unsigned long) be64toh(request.session_id)), &iter);
-       node = lttng_ht_iter_get_node_ulong(&iter);
-       if (node == NULL) {
+       session = session_find_by_id(sessions_ht, be64toh(request.session_id));
+       if (!session) {
                DBG("Relay session %" PRIu64 " not found",
                                be64toh(request.session_id));
                response.status = htobe32(VIEWER_ATTACH_UNK);
                goto send_reply;
        }
 
-       session = caa_container_of(node, struct relay_session, session_n);
        if (cmd->session_id == session->id) {
                /* Same viewer already attached, just send the stream list. */
                send_streams = 1;
@@ -1093,10 +950,8 @@ int viewer_attach_session(struct relay_command *cmd,
 
        switch (be32toh(request.seek)) {
        case VIEWER_SEEK_BEGINNING:
-               /* Default behaviour. */
-               break;
        case VIEWER_SEEK_LAST:
-               seek_last = 1;
+               seek_type = be32toh(request.seek);
                break;
        default:
                ERR("Wrong seek parameter");
@@ -1105,58 +960,20 @@ int viewer_attach_session(struct relay_command *cmd,
                goto send_reply;
        }
 
-       if (send_streams) {
-               /* We should only be there if we have a session to attach to. */
-               assert(session);
-
-               /*
-                * Fill the viewer_streams_ht to count the number of streams
-                * ready to be sent and avoid concurrency issues on the
-                * relay_streams_ht and don't rely on a total session stream count.
-                */
-               pthread_mutex_lock(&session->viewer_ready_lock);
-               cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, node, node) {
-                       struct relay_viewer_stream *vstream;
-
-                       health_code_update();
-
-                       node = lttng_ht_iter_get_node_ulong(&iter);
-                       if (!node) {
-                               continue;
-                       }
-                       stream = caa_container_of(node, struct relay_stream, stream_n);
-                       if (stream->session != cmd->session) {
-                               continue;
-                       }
-
-                       /*
-                        * Don't send streams with no ctf_trace, they are not
-                        * ready to be read.
-                        */
-                       if (!stream->ctf_trace || !stream->viewer_ready) {
-                               continue;
-                       }
-
-                       vstream = live_find_viewer_stream_by_id(stream->stream_handle);
-                       if (!vstream) {
-                               ret = init_viewer_stream(stream, seek_last);
-                               if (ret < 0) {
-                                       pthread_mutex_unlock(&session->viewer_ready_lock);
-                                       goto end_unlock;
-                               }
-                       }
-                       nb_streams++;
-               }
-               pthread_mutex_unlock(&session->viewer_ready_lock);
+       if (!send_streams) {
+               goto send_reply;
+       }
 
-               response.streams_count = htobe32(nb_streams);
+       ret = make_viewer_streams(session, seek_type, &nb_streams, NULL, NULL);
+       if (ret < 0) {
+               goto end_unlock;
        }
+       response.streams_count = htobe32(nb_streams);
 
 send_reply:
        health_code_update();
-       ret = cmd->sock->ops->sendmsg(cmd->sock, &response, sizeof(response), 0);
+       ret = send_response(cmd->sock, &response, sizeof(response));
        if (ret < 0) {
-               ERR("Relay sending viewer attach response");
                goto end_unlock;
        }
        health_code_update();
@@ -1170,167 +987,14 @@ send_reply:
                goto end_unlock;
        }
 
-       /* We should only be there if we have a session to attach to. */
-       assert(session);
-       cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, node, node) {
-               health_code_update();
-
-               node64 = lttng_ht_iter_get_node_u64(&iter);
-               if (!node64) {
-                       continue;
-               }
-               viewer_stream = caa_container_of(node64, struct relay_viewer_stream,
-                               stream_n);
-               if (viewer_stream->session_id != cmd->session->id) {
-                       continue;
-               }
-
-               send_stream.id = htobe64(viewer_stream->stream_handle);
-               send_stream.ctf_trace_id = htobe64(viewer_stream->ctf_trace->id);
-               send_stream.metadata_flag = htobe32(viewer_stream->metadata_flag);
-               strncpy(send_stream.path_name, viewer_stream->path_name,
-                               sizeof(send_stream.path_name));
-               strncpy(send_stream.channel_name, viewer_stream->channel_name,
-                               sizeof(send_stream.channel_name));
-
-               ret = cmd->sock->ops->sendmsg(cmd->sock, &send_stream,
-                               sizeof(send_stream), 0);
-               if (ret < 0) {
-                       ERR("Relay sending stream %" PRIu64, viewer_stream->stream_handle);
-                       goto end_unlock;
-               }
-               DBG("Sent stream %" PRIu64 " to viewer", viewer_stream->stream_handle);
-               viewer_stream->sent_flag = 1;
+       /* Send stream and ignore the sent flag. */
+       ret = send_viewer_streams(cmd->sock, session, 1);
+       if (ret < 0) {
+               goto end_unlock;
        }
-       ret = 0;
 
 end_unlock:
        rcu_read_unlock();
-end_no_session:
-error:
-       return ret;
-}
-
-/*
- * Get viewer stream from stream id.
- *
- * RCU read side lock MUST be acquired.
- */
-struct relay_viewer_stream *live_find_viewer_stream_by_id(uint64_t stream_id)
-{
-       struct lttng_ht_node_u64 *node;
-       struct lttng_ht_iter iter;
-       struct relay_viewer_stream *stream = NULL;
-
-       lttng_ht_lookup(viewer_streams_ht, &stream_id, &iter);
-       node = lttng_ht_iter_get_node_u64(&iter);
-       if (node == NULL) {
-               DBG("Relay viewer stream %" PRIu64 " not found", stream_id);
-               goto end;
-       }
-       stream = caa_container_of(node, struct relay_viewer_stream, stream_n);
-
-end:
-       return stream;
-}
-
-static
-void deferred_free_viewer_stream(struct rcu_head *head)
-{
-       struct relay_viewer_stream *stream =
-               caa_container_of(head, struct relay_viewer_stream, rcu_node);
-
-       free(stream->path_name);
-       free(stream->channel_name);
-       free(stream);
-}
-
-static
-void delete_viewer_stream(struct relay_viewer_stream *vstream)
-{
-       int delret;
-       struct lttng_ht_iter iter;
-
-       iter.iter.node = &vstream->stream_n.node;
-       delret = lttng_ht_del(viewer_streams_ht, &iter);
-       assert(!delret);
-}
-
-static
-void destroy_viewer_stream(struct relay_viewer_stream *vstream)
-{
-       unsigned long ret_ref;
-       int ret;
-
-       assert(vstream);
-       ret_ref = uatomic_add_return(&vstream->ctf_trace->refcount, -1);
-       assert(ret_ref >= 0);
-
-       if (vstream->read_fd >= 0) {
-               ret = close(vstream->read_fd);
-               if (ret < 0) {
-                       PERROR("close read_fd");
-               }
-       }
-       if (vstream->index_read_fd >= 0) {
-               ret = close(vstream->index_read_fd);
-               if (ret < 0) {
-                       PERROR("close index_read_fd");
-               }
-       }
-
-       /*
-        * If the only stream left in the HT is the metadata stream,
-        * we need to remove it because we won't detect a EOF for this
-        * stream.
-        */
-       if (ret_ref == 1 && vstream->ctf_trace->viewer_metadata_stream) {
-               delete_viewer_stream(vstream->ctf_trace->viewer_metadata_stream);
-               destroy_viewer_stream(vstream->ctf_trace->viewer_metadata_stream);
-               vstream->ctf_trace->metadata_stream = NULL;
-               DBG("Freeing ctf_trace %" PRIu64, vstream->ctf_trace->id);
-               /*
-                * The streaming-side is already closed and we can't receive a new
-                * stream concurrently at this point (since the session is being
-                * destroyed), so when we detect the refcount equals 0, we are the
-                * only owners of the ctf_trace and we can free it ourself.
-                */
-               free(vstream->ctf_trace);
-       }
-
-       call_rcu(&vstream->rcu_node, deferred_free_viewer_stream);
-}
-
-/*
- * Atomically check if new streams got added in the session since the last
- * check and reset the flag to 0.
- *
- * Returns 1 if new streams got added, 0 if nothing changed, a negative value
- * on error.
- */
-static
-int check_new_streams(uint64_t session_id, struct lttng_ht *sessions_ht)
-{
-       struct lttng_ht_node_ulong *node;
-       struct lttng_ht_iter iter;
-       struct relay_session *session;
-       unsigned long current_val;
-       int ret;
-
-       lttng_ht_lookup(sessions_ht,
-                       (void *)((unsigned long) session_id), &iter);
-       node = lttng_ht_iter_get_node_ulong(&iter);
-       if (node == NULL) {
-               DBG("Relay session %" PRIu64 " not found", session_id);
-               ret = -1;
-               goto error;
-       }
-
-       session = caa_container_of(node, struct relay_session, session_n);
-
-       current_val = uatomic_cmpxchg(&session->new_streams, 1, 0);
-       ret = current_val;
-
 error:
        return ret;
 }
@@ -1356,24 +1020,16 @@ int viewer_get_next_index(struct relay_command *cmd,
 
        DBG("Viewer get next index");
 
-       if (cmd->version_check_done == 0) {
-               ERR("Trying to request index before version check");
-               ret = -1;
-               goto end_no_session;
-       }
-
        health_code_update();
-       ret = cmd->sock->ops->recvmsg(cmd->sock, &request_index,
-                       sizeof(request_index), 0);
-       if (ret < 0 || ret != sizeof(request_index)) {
-               ret = -1;
-               ERR("Relay didn't receive the whole packet");
+
+       ret = recv_request(cmd->sock, &request_index, sizeof(request_index));
+       if (ret < 0) {
                goto end;
        }
        health_code_update();
 
        rcu_read_lock();
-       vstream = live_find_viewer_stream_by_id(be64toh(request_index.stream_id));
+       vstream = viewer_stream_find_by_id(be64toh(request_index.stream_id));
        if (!vstream) {
                ret = -1;
                goto end_unlock;
@@ -1391,7 +1047,8 @@ int viewer_get_next_index(struct relay_command *cmd,
 
        /* First time, we open the index file */
        if (vstream->index_read_fd < 0) {
-               ret = open_index(vstream);
+               ret = index_open(vstream->path_name, vstream->channel_name,
+                               vstream->tracefile_count, vstream->tracefile_count_current);
                if (ret == -ENOENT) {
                        /*
                         * The index is created only when the first data packet arrives, it
@@ -1403,6 +1060,7 @@ int viewer_get_next_index(struct relay_command *cmd,
                        viewer_index.status = htobe32(VIEWER_INDEX_ERR);
                        goto send_reply;
                }
+               vstream->index_read_fd = ret;
        }
 
        rstream = relay_stream_find_by_id(vstream->stream_handle);
@@ -1410,13 +1068,13 @@ int viewer_get_next_index(struct relay_command *cmd,
                if (vstream->abort_flag) {
                        /* Rotate on abort (overwrite). */
                        DBG("Viewer rotate because of overwrite");
-                       ret = rotate_viewer_stream(vstream, rstream);
+                       ret = viewer_stream_rotate(vstream, rstream);
                        if (ret < 0) {
                                goto end_unlock;
                        } else if (ret == 1) {
                                viewer_index.status = htobe32(VIEWER_INDEX_HUP);
-                               delete_viewer_stream(vstream);
-                               destroy_viewer_stream(vstream);
+                               viewer_stream_delete(vstream);
+                               viewer_stream_destroy(vstream);
                                goto send_reply;
                        }
                }
@@ -1446,8 +1104,8 @@ int viewer_get_next_index(struct relay_command *cmd,
                        vstream->total_index_received == vstream->last_sent_index) {
                /* Last index sent and current tracefile closed in write */
                viewer_index.status = htobe32(VIEWER_INDEX_HUP);
-               delete_viewer_stream(vstream);
-               destroy_viewer_stream(vstream);
+               viewer_stream_delete(vstream);
+               viewer_stream_destroy(vstream);
                goto send_reply;
        } else {
                vstream->close_write_flag = 1;
@@ -1474,17 +1132,18 @@ int viewer_get_next_index(struct relay_command *cmd,
                 */
                viewer_index.status = htobe32(VIEWER_INDEX_RETRY);
                pthread_mutex_unlock(&vstream->overwrite_lock);
-               ret = rotate_viewer_stream(vstream, rstream);
+               ret = viewer_stream_rotate(vstream, rstream);
                if (ret < 0) {
                        goto end_unlock;
                } else if (ret == 1) {
                        viewer_index.status = htobe32(VIEWER_INDEX_HUP);
-                       delete_viewer_stream(vstream);
-                       destroy_viewer_stream(vstream);
+                       viewer_stream_delete(vstream);
+                       viewer_stream_destroy(vstream);
                        goto send_reply;
                }
                goto send_reply;
        }
+
        ret = lttng_read(vstream->index_read_fd, &packet_index,
                        sizeof(packet_index));
        pthread_mutex_unlock(&vstream->overwrite_lock);
@@ -1495,18 +1154,17 @@ int viewer_get_next_index(struct relay_command *cmd,
                if (vstream->close_write_flag == 1) {
                        viewer_index.status = htobe32(VIEWER_INDEX_RETRY);
                        /* Rotate on normal EOF */
-                       ret = rotate_viewer_stream(vstream, rstream);
+                       ret = viewer_stream_rotate(vstream, rstream);
                        if (ret < 0) {
                                goto end_unlock;
                        } else if (ret == 1) {
                                viewer_index.status = htobe32(VIEWER_INDEX_HUP);
-                               delete_viewer_stream(vstream);
-                               destroy_viewer_stream(vstream);
+                               viewer_stream_delete(vstream);
+                               viewer_stream_destroy(vstream);
                                goto send_reply;
                        }
                } else {
-                       PERROR("Relay reading index file %d",
-                                       vstream->index_read_fd);
+                       PERROR("Relay reading index file %d", vstream->index_read_fd);
                        viewer_index.status = htobe32(VIEWER_INDEX_ERR);
                }
                goto send_reply;
@@ -1529,21 +1187,19 @@ int viewer_get_next_index(struct relay_command *cmd,
 send_reply:
        viewer_index.flags = htobe32(viewer_index.flags);
        health_code_update();
-       ret = cmd->sock->ops->sendmsg(cmd->sock, &viewer_index,
-                       sizeof(viewer_index), 0);
+
+       ret = send_response(cmd->sock, &viewer_index, sizeof(viewer_index));
        if (ret < 0) {
-               ERR("Relay index to viewer");
                goto end_unlock;
        }
        health_code_update();
 
-       DBG("Index %" PRIu64 "for stream %" PRIu64 "sent",
+       DBG("Index %" PRIu64 " for stream %" PRIu64 " sent",
                        vstream->last_sent_index, vstream->stream_handle);
 
 end_unlock:
        rcu_read_unlock();
 
-end_no_session:
 end:
        return ret;
 }
@@ -1569,18 +1225,10 @@ int viewer_get_packet(struct relay_command *cmd,
 
        DBG2("Relay get data packet");
 
-       if (cmd->version_check_done == 0) {
-               ERR("Trying to get packet before version check");
-               ret = -1;
-               goto end;
-       }
-
        health_code_update();
-       ret = cmd->sock->ops->recvmsg(cmd->sock, &get_packet_info,
-                       sizeof(get_packet_info), 0);
-       if (ret < 0 || ret != sizeof(get_packet_info)) {
-               ret = -1;
-               ERR("Relay didn't receive the whole packet");
+
+       ret = recv_request(cmd->sock, &get_packet_info, sizeof(get_packet_info));
+       if (ret < 0) {
                goto end;
        }
        health_code_update();
@@ -1589,7 +1237,7 @@ int viewer_get_packet(struct relay_command *cmd,
        memset(&reply, 0, sizeof(reply));
 
        rcu_read_lock();
-       stream = live_find_viewer_stream_by_id(be64toh(get_packet_info.stream_id));
+       stream = viewer_stream_find_by_id(be64toh(get_packet_info.stream_id));
        if (!stream) {
                goto error;
        }
@@ -1687,18 +1335,17 @@ send_reply:
        reply.flags = htobe32(reply.flags);
 
        health_code_update();
-       ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
+
+       ret = send_response(cmd->sock, &reply, sizeof(reply));
        if (ret < 0) {
-               ERR("Relay data header to viewer");
                goto end_unlock;
        }
        health_code_update();
 
        if (send_data) {
                health_code_update();
-               ret = cmd->sock->ops->sendmsg(cmd->sock, data, len, 0);
+               ret = send_response(cmd->sock, data, len);
                if (ret < 0) {
-                       ERR("Relay send data to viewer");
                        goto end_unlock;
                }
                health_code_update();
@@ -1735,24 +1382,16 @@ int viewer_get_metadata(struct relay_command *cmd)
 
        DBG("Relay get metadata");
 
-       if (cmd->version_check_done == 0) {
-               ERR("Trying to get metadata before version check");
-               ret = -1;
-               goto end;
-       }
-
        health_code_update();
-       ret = cmd->sock->ops->recvmsg(cmd->sock, &request,
-                       sizeof(request), 0);
-       if (ret < 0 || ret != sizeof(request)) {
-               ret = -1;
-               ERR("Relay didn't receive the whole packet");
+
+       ret = recv_request(cmd->sock, &request, sizeof(request));
+       if (ret < 0) {
                goto end;
        }
        health_code_update();
 
        rcu_read_lock();
-       stream = live_find_viewer_stream_by_id(be64toh(request.stream_id));
+       stream = viewer_stream_find_by_id(be64toh(request.stream_id));
        if (!stream || !stream->metadata_flag) {
                ERR("Invalid metadata stream");
                goto error;
@@ -1806,17 +1445,15 @@ error:
 
 send_reply:
        health_code_update();
-       ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
+       ret = send_response(cmd->sock, &reply, sizeof(reply));
        if (ret < 0) {
-               ERR("Relay data header to viewer");
                goto end_unlock;
        }
        health_code_update();
 
        if (len > 0) {
-               ret = cmd->sock->ops->sendmsg(cmd->sock, data, len, 0);
+               ret = send_response(cmd->sock, data, len);
                if (ret < 0) {
-                       ERR("Relay send data to viewer");
                        goto end_unlock;
                }
        }
@@ -1840,14 +1477,9 @@ static
 void live_relay_unknown_command(struct relay_command *cmd)
 {
        struct lttcomm_relayd_generic_reply reply;
-       int ret;
 
        reply.ret_code = htobe32(LTTNG_ERR_UNK);
-       ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
-                       sizeof(struct lttcomm_relayd_generic_reply), 0);
-       if (ret < 0) {
-               ERR("Relay sending unknown command");
-       }
+       (void) send_response(cmd->sock, &reply, sizeof(reply));
 }
 
 /*
@@ -1858,8 +1490,25 @@ int process_control(struct lttng_viewer_cmd *recv_hdr,
                struct relay_command *cmd, struct lttng_ht *sessions_ht)
 {
        int ret = 0;
+       uint32_t msg_value;
+
+       assert(recv_hdr);
+       assert(cmd);
+       assert(sessions_ht);
+
+       msg_value = be32toh(recv_hdr->cmd);
+
+       /*
+        * Make sure we've done the version check before any command other then a
+        * new client connection.
+        */
+       if (msg_value != VIEWER_CONNECT && !cmd->version_check_done) {
+               ERR("Viewer cmd value %" PRIu32 " before version check", msg_value);
+               ret = -1;
+               goto end;
+       }
 
-       switch (be32toh(recv_hdr->cmd)) {
+       switch (msg_value) {
        case VIEWER_CONNECT:
                ret = viewer_connect(cmd);
                break;
@@ -1982,7 +1631,7 @@ void viewer_del_streams(uint64_t session_id)
                        continue;
                }
 
-               delete_viewer_stream(stream);
+               viewer_stream_delete(stream);
                assert(stream->ctf_trace);
 
                if (stream->metadata_flag) {
@@ -1994,7 +1643,7 @@ void viewer_del_streams(uint64_t session_id)
                        stream->ctf_trace->metadata_sent = 0;
                        stream->ctf_trace->viewer_metadata_stream = NULL;
                } else {
-                       destroy_viewer_stream(stream);
+                       viewer_stream_destroy(stream);
                }
        }
        rcu_read_unlock();
index b5e7c75fdd4b901140c9ec91088791cd1b9f599b..6dfc803236885bb1848f5889b75bf3207b908732 100644 (file)
@@ -29,6 +29,7 @@
 #include <common/index/ctf-index.h>
 
 #include "ctf-trace.h"
+#include "session.h"
 
 /*
  * Queue used to enqueue relay requests
@@ -55,48 +56,6 @@ struct relay_stream_recv_handle {
        struct cds_list_head node;
 };
 
-/*
- * Represents a session for the relay point of view
- */
-struct relay_session {
-       /*
-        * This session id is used to identify a set of stream to a tracing session
-        * but also make sure we have a unique session id associated with a session
-        * daemon which can provide multiple data source.
-        */
-       uint64_t id;
-       struct lttcomm_sock *sock;
-       char session_name[NAME_MAX];
-       char hostname[HOST_NAME_MAX];
-       uint32_t live_timer;
-       struct lttng_ht_node_ulong session_n;
-       struct rcu_head rcu_node;
-       uint32_t viewer_attached;
-       uint32_t stream_count;
-       /* Tell if this session is for a snapshot or not. */
-       unsigned int snapshot:1;
-
-       /*
-        * Indicate version protocol for this session. This is especially useful
-        * for the data thread that has no idea which version it operates on since
-        * linking control/data sockets is non trivial.
-        */
-       uint64_t minor;
-       uint64_t major;
-       /*
-        * Flag checked and exchanged with uatomic_cmpxchg to tell the
-        * viewer-side if new streams got added since the last check.
-        */
-       unsigned long new_streams;
-
-       /*
-        * Used to synchronize the process where we flag every streams readiness
-        * for the viewer when the streams_sent message is received and the viewer
-        * process of sending those streams.
-        */
-       pthread_mutex_t viewer_ready_lock;
-};
-
 /*
  * Represents a stream in the relay
  */
@@ -123,7 +82,6 @@ struct relay_stream {
        uint64_t oldest_tracefile_id;
 
        uint64_t total_index_received;
-       struct relay_viewer_stream *viewer_stream;
        uint64_t last_net_seq_num;
 
        /*
@@ -174,51 +132,6 @@ struct relay_stream {
        unsigned int viewer_ready:1;
 };
 
-/*
- * Shadow copy of the relay_stream structure for the viewer side.  The only
- * fields updated by the writer (streaming side) after allocation are :
- * total_index_received and close_flag. Everything else is updated by the
- * reader (viewer side).
- */
-struct relay_viewer_stream {
-       uint64_t stream_handle;
-       uint64_t session_id;
-       int read_fd;
-       int index_read_fd;
-       char *path_name;
-       char *channel_name;
-       uint64_t last_sent_index;
-       uint64_t total_index_received;
-       uint64_t tracefile_count;
-       uint64_t tracefile_count_current;
-       /* Stop after reading this tracefile. */
-       uint64_t tracefile_count_last;
-       struct lttng_ht_node_u64 stream_n;
-       struct rcu_head rcu_node;
-       struct ctf_trace *ctf_trace;
-       /*
-        * This lock blocks only when the writer is about to start overwriting
-        * a file currently read by the reader.
-        *
-        * This is nested INSIDE the viewer_stream_rotation_lock.
-        */
-       pthread_mutex_t overwrite_lock;
-       /* Information telling us if the stream is a metadata stream. */
-       unsigned int metadata_flag:1;
-       /*
-        * Information telling us that the stream is closed in write, so
-        * we don't expect new indexes and we can read up to EOF.
-        */
-       unsigned int close_write_flag:1;
-       /*
-        * If the streaming side closes a FD in use in the viewer side,
-        * it sets this flag to inform that it is a normal error.
-        */
-       unsigned int abort_flag:1;
-       /* Indicates if this stream has been sent to a viewer client. */
-       unsigned int sent_flag:1;
-};
-
 /*
  * Internal structure to map a socket with the corresponding session.
  * A hashtable indexed on the socket FD is used for the lookups.
diff --git a/src/bin/lttng-relayd/lttng-viewer-abi.h b/src/bin/lttng-relayd/lttng-viewer-abi.h
new file mode 100644 (file)
index 0000000..1f6955b
--- /dev/null
@@ -0,0 +1,224 @@
+#ifndef LTTNG_VIEWER_ABI_H
+#define LTTNG_VIEWER_ABI_H
+
+/*
+ * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
+ *                      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ *                      David Goulet <dgoulet@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.
+ */
+
+#include <limits.h>
+
+#define LTTNG_VIEWER_PATH_MAX          4096
+#define LTTNG_VIEWER_NAME_MAX          255
+#define LTTNG_VIEWER_ABI_HOST_NAME_MAX 64
+
+/* Flags in reply to get_next_index and get_packet. */
+/* New metadata is required to read this packet. */
+#define LTTNG_VIEWER_FLAG_NEW_METADATA (1 << 0)
+/* New stream got added to the trace. */
+#define LTTNG_VIEWER_FLAG_NEW_STREAM   (1 << 1)
+
+enum lttng_viewer_command {
+       VIEWER_CONNECT          = 1,
+       VIEWER_LIST_SESSIONS    = 2,
+       VIEWER_ATTACH_SESSION   = 3,
+       VIEWER_GET_NEXT_INDEX   = 4,
+       VIEWER_GET_PACKET       = 5,
+       VIEWER_GET_METADATA     = 6,
+       VIEWER_GET_NEW_STREAMS  = 7,
+};
+
+enum lttng_viewer_attach_return_code {
+       VIEWER_ATTACH_OK        = 1, /* If the attach command succeeded. */
+       VIEWER_ATTACH_ALREADY   = 2, /* If a viewer is already attached. */
+       VIEWER_ATTACH_UNK       = 3, /* If the session ID is unknown. */
+       VIEWER_ATTACH_NOT_LIVE  = 4, /* If the session is not live. */
+       VIEWER_ATTACH_SEEK_ERR  = 5, /* Seek error. */
+};
+
+enum lttng_viewer_next_index_return_code {
+       VIEWER_INDEX_OK         = 1, /* Index is available. */
+       VIEWER_INDEX_RETRY      = 2, /* Index not yet available. */
+       VIEWER_INDEX_HUP        = 3, /* Index closed (trace destroyed). */
+       VIEWER_INDEX_ERR        = 4, /* Unknow error. */
+       VIEWER_INDEX_INACTIVE   = 5, /* Inactive stream beacon. */
+       VIEWER_INDEX_EOF        = 6, /* End of index file. */
+};
+
+enum lttng_viewer_get_packet_return_code {
+       VIEWER_GET_PACKET_OK            = 1,
+       VIEWER_GET_PACKET_RETRY         = 2,
+       VIEWER_GET_PACKET_ERR           = 3,
+       VIEWER_GET_PACKET_EOF           = 4,
+};
+
+enum lttng_viewer_get_metadata_return_code {
+       VIEWER_METADATA_OK      = 1,
+       VIEWER_NO_NEW_METADATA  = 2,
+       VIEWER_METADATA_ERR     = 3,
+};
+
+enum lttng_viewer_connection_type {
+       VIEWER_CLIENT_COMMAND           = 1,
+       VIEWER_CLIENT_NOTIFICATION      = 2,
+};
+
+enum lttng_viewer_seek {
+       /* Receive the trace packets from the beginning. */
+       VIEWER_SEEK_BEGINNING   = 1,
+       /* Receive the trace packets from now. */
+       VIEWER_SEEK_LAST        = 2,
+};
+
+enum lttng_viewer_new_streams_return_code {
+       VIEWER_NEW_STREAMS_OK           = 1, /* If new streams are being sent. */
+       VIEWER_NEW_STREAMS_NO_NEW       = 2, /* If no new streams are available. */
+       VIEWER_NEW_STREAMS_ERR          = 3, /* Error. */
+};
+
+struct lttng_viewer_session {
+       uint64_t id;
+       uint32_t live_timer;
+       uint32_t clients;
+       uint32_t streams;
+       char hostname[LTTNG_VIEWER_ABI_HOST_NAME_MAX];
+       char session_name[LTTNG_VIEWER_NAME_MAX];
+} __attribute__((__packed__));
+
+struct lttng_viewer_stream {
+       uint64_t id;
+       uint64_t ctf_trace_id;
+       int metadata_flag;
+       char path_name[LTTNG_VIEWER_PATH_MAX];
+       char channel_name[LTTNG_VIEWER_NAME_MAX];
+} __attribute__((__packed__));
+
+struct lttng_viewer_cmd {
+       uint64_t data_size;     /* data size following this header */
+       uint32_t cmd;           /* enum lttcomm_relayd_command */
+       uint32_t cmd_version;   /* command version */
+} __attribute__((__packed__));
+
+/*
+ * CONNECT payload.
+ */
+struct lttng_viewer_connect {
+       /* session ID assigned by the relay for command connections */
+       uint64_t viewer_session_id;
+       uint32_t major;
+       uint32_t minor;
+       uint32_t type; /* enum lttng_viewer_connection_type */
+} __attribute__((__packed__));
+
+/*
+ * VIEWER_LIST_SESSIONS payload.
+ */
+struct lttng_viewer_list_sessions {
+       uint32_t sessions_count;
+       char session_list[];            /* struct lttng_viewer_session */
+} __attribute__((__packed__));
+
+/*
+ * VIEWER_ATTACH_SESSION payload.
+ */
+struct lttng_viewer_attach_session_request {
+       uint64_t session_id;
+       uint64_t offset;        /* unused for now */
+       uint32_t seek;          /* enum lttng_viewer_seek */
+} __attribute__((__packed__));
+
+struct lttng_viewer_attach_session_response {
+       /* enum lttng_viewer_attach_return_code */
+       uint32_t status;
+       uint32_t streams_count;
+       /* struct lttng_viewer_stream */
+       char stream_list[];
+} __attribute__((__packed__));
+
+/*
+ * VIEWER_GET_NEXT_INDEX payload.
+ */
+struct lttng_viewer_get_next_index {
+       uint64_t stream_id;
+} __attribute__ ((__packed__));
+
+struct lttng_viewer_index {
+       uint64_t offset;
+       uint64_t packet_size;
+       uint64_t content_size;
+       uint64_t timestamp_begin;
+       uint64_t timestamp_end;
+       uint64_t events_discarded;
+       uint64_t stream_id;
+       /* enum lttng_viewer_next_index_return_code */
+       uint32_t status;
+       uint32_t flags; /* LTTNG_VIEWER_FLAG_* */
+} __attribute__ ((__packed__));
+
+/*
+ * VIEWER_GET_PACKET payload.
+ */
+struct lttng_viewer_get_packet {
+       uint64_t stream_id;
+       uint64_t offset;
+       uint32_t len;
+} __attribute__((__packed__));
+
+struct lttng_viewer_trace_packet {
+       /* enum lttng_viewer_get_packet_return_code */
+       uint32_t status;
+       uint32_t len;
+       uint32_t flags; /* LTTNG_VIEWER_FLAG_* */
+       char data[];
+} __attribute__((__packed__));
+
+/*
+ * VIEWER_GET_METADATA payload.
+ */
+struct lttng_viewer_get_metadata {
+       uint64_t stream_id;
+} __attribute__((__packed__));
+
+struct lttng_viewer_metadata_packet {
+       uint64_t len;
+       /* enum lttng_viewer_get_metadata_return_code */
+       uint32_t status;
+       char data[];
+} __attribute__((__packed__));
+
+/*
+ * VIEWER_GET_NEW_STREAMS payload.
+ */
+struct lttng_viewer_new_streams_request {
+       uint64_t session_id;
+} __attribute__((__packed__));
+
+struct lttng_viewer_new_streams_response {
+       /* enum lttng_viewer_new_streams_return_code */
+       uint32_t status;
+       uint32_t streams_count;
+       /* struct lttng_viewer_stream */
+       char stream_list[];
+} __attribute__((__packed__));
+
+
+#endif /* LTTNG_VIEWER_ABI_H */
diff --git a/src/bin/lttng-relayd/lttng-viewer.h b/src/bin/lttng-relayd/lttng-viewer.h
deleted file mode 100644 (file)
index baa8964..0000000
+++ /dev/null
@@ -1,224 +0,0 @@
-#ifndef LTTNG_VIEWER_H
-#define LTTNG_VIEWER_H
-
-/*
- * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
- *                      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
- *                      David Goulet <dgoulet@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.
- */
-
-#include <limits.h>
-
-#define LTTNG_VIEWER_PATH_MAX          4096
-#define LTTNG_VIEWER_NAME_MAX          255
-#define LTTNG_VIEWER_HOST_NAME_MAX     64
-
-/* Flags in reply to get_next_index and get_packet. */
-/* New metadata is required to read this packet. */
-#define LTTNG_VIEWER_FLAG_NEW_METADATA (1 << 0)
-/* New stream got added to the trace. */
-#define LTTNG_VIEWER_FLAG_NEW_STREAM   (1 << 1)
-
-enum lttng_viewer_command {
-       VIEWER_CONNECT          = 1,
-       VIEWER_LIST_SESSIONS    = 2,
-       VIEWER_ATTACH_SESSION   = 3,
-       VIEWER_GET_NEXT_INDEX   = 4,
-       VIEWER_GET_PACKET       = 5,
-       VIEWER_GET_METADATA     = 6,
-       VIEWER_GET_NEW_STREAMS  = 7,
-};
-
-enum lttng_viewer_attach_return_code {
-       VIEWER_ATTACH_OK        = 1, /* If the attach command succeeded. */
-       VIEWER_ATTACH_ALREADY   = 2, /* If a viewer is already attached. */
-       VIEWER_ATTACH_UNK       = 3, /* If the session ID is unknown. */
-       VIEWER_ATTACH_NOT_LIVE  = 4, /* If the session is not live. */
-       VIEWER_ATTACH_SEEK_ERR  = 5, /* Seek error. */
-};
-
-enum lttng_viewer_next_index_return_code {
-       VIEWER_INDEX_OK         = 1, /* Index is available. */
-       VIEWER_INDEX_RETRY      = 2, /* Index not yet available. */
-       VIEWER_INDEX_HUP        = 3, /* Index closed (trace destroyed). */
-       VIEWER_INDEX_ERR        = 4, /* Unknow error. */
-       VIEWER_INDEX_INACTIVE   = 5, /* Inactive stream beacon. */
-       VIEWER_INDEX_EOF        = 6, /* End of index file. */
-};
-
-enum lttng_viewer_get_packet_return_code {
-       VIEWER_GET_PACKET_OK            = 1,
-       VIEWER_GET_PACKET_RETRY         = 2,
-       VIEWER_GET_PACKET_ERR           = 3,
-       VIEWER_GET_PACKET_EOF           = 4,
-};
-
-enum lttng_viewer_get_metadata_return_code {
-       VIEWER_METADATA_OK      = 1,
-       VIEWER_NO_NEW_METADATA  = 2,
-       VIEWER_METADATA_ERR     = 3,
-};
-
-enum lttng_viewer_connection_type {
-       VIEWER_CLIENT_COMMAND           = 1,
-       VIEWER_CLIENT_NOTIFICATION      = 2,
-};
-
-enum lttng_viewer_seek {
-       /* Receive the trace packets from the beginning. */
-       VIEWER_SEEK_BEGINNING   = 1,
-       /* Receive the trace packets from now. */
-       VIEWER_SEEK_LAST        = 2,
-};
-
-enum lttng_viewer_new_streams_return_code {
-       VIEWER_NEW_STREAMS_OK           = 1, /* If new streams are being sent. */
-       VIEWER_NEW_STREAMS_NO_NEW       = 2, /* If no new streams are available. */
-       VIEWER_NEW_STREAMS_ERR          = 3, /* Error. */
-};
-
-struct lttng_viewer_session {
-       uint64_t id;
-       uint32_t live_timer;
-       uint32_t clients;
-       uint32_t streams;
-       char hostname[LTTNG_VIEWER_HOST_NAME_MAX];
-       char session_name[LTTNG_VIEWER_NAME_MAX];
-} __attribute__((__packed__));
-
-struct lttng_viewer_stream {
-       uint64_t id;
-       uint64_t ctf_trace_id;
-       int metadata_flag;
-       char path_name[LTTNG_VIEWER_PATH_MAX];
-       char channel_name[LTTNG_VIEWER_NAME_MAX];
-} __attribute__((__packed__));
-
-struct lttng_viewer_cmd {
-       uint64_t data_size;     /* data size following this header */
-       uint32_t cmd;           /* enum lttcomm_relayd_command */
-       uint32_t cmd_version;   /* command version */
-} __attribute__((__packed__));
-
-/*
- * CONNECT payload.
- */
-struct lttng_viewer_connect {
-       /* session ID assigned by the relay for command connections */
-       uint64_t viewer_session_id;
-       uint32_t major;
-       uint32_t minor;
-       uint32_t type; /* enum lttng_viewer_connection_type */
-} __attribute__((__packed__));
-
-/*
- * VIEWER_LIST_SESSIONS payload.
- */
-struct lttng_viewer_list_sessions {
-       uint32_t sessions_count;
-       char session_list[];            /* struct lttng_viewer_session */
-} __attribute__((__packed__));
-
-/*
- * VIEWER_ATTACH_SESSION payload.
- */
-struct lttng_viewer_attach_session_request {
-       uint64_t session_id;
-       uint64_t offset;        /* unused for now */
-       uint32_t seek;          /* enum lttng_viewer_seek */
-} __attribute__((__packed__));
-
-struct lttng_viewer_attach_session_response {
-       /* enum lttng_viewer_attach_return_code */
-       uint32_t status;
-       uint32_t streams_count;
-       /* struct lttng_viewer_stream */
-       char stream_list[];
-} __attribute__((__packed__));
-
-/*
- * VIEWER_GET_NEXT_INDEX payload.
- */
-struct lttng_viewer_get_next_index {
-       uint64_t stream_id;
-} __attribute__ ((__packed__));
-
-struct lttng_viewer_index {
-       uint64_t offset;
-       uint64_t packet_size;
-       uint64_t content_size;
-       uint64_t timestamp_begin;
-       uint64_t timestamp_end;
-       uint64_t events_discarded;
-       uint64_t stream_id;
-       /* enum lttng_viewer_next_index_return_code */
-       uint32_t status;
-       uint32_t flags; /* LTTNG_VIEWER_FLAG_* */
-} __attribute__ ((__packed__));
-
-/*
- * VIEWER_GET_PACKET payload.
- */
-struct lttng_viewer_get_packet {
-       uint64_t stream_id;
-       uint64_t offset;
-       uint32_t len;
-} __attribute__((__packed__));
-
-struct lttng_viewer_trace_packet {
-       /* enum lttng_viewer_get_packet_return_code */
-       uint32_t status;
-       uint32_t len;
-       uint32_t flags; /* LTTNG_VIEWER_FLAG_* */
-       char data[];
-} __attribute__((__packed__));
-
-/*
- * VIEWER_GET_METADATA payload.
- */
-struct lttng_viewer_get_metadata {
-       uint64_t stream_id;
-} __attribute__((__packed__));
-
-struct lttng_viewer_metadata_packet {
-       uint64_t len;
-       /* enum lttng_viewer_get_metadata_return_code */
-       uint32_t status;
-       char data[];
-} __attribute__((__packed__));
-
-/*
- * VIEWER_GET_NEW_STREAMS payload.
- */
-struct lttng_viewer_new_streams_request {
-       uint64_t session_id;
-} __attribute__((__packed__));
-
-struct lttng_viewer_new_streams_response {
-       /* enum lttng_viewer_new_streams_return_code */
-       uint32_t status;
-       uint32_t streams_count;
-       /* struct lttng_viewer_stream */
-       char stream_list[];
-} __attribute__((__packed__));
-
-
-#endif /* LTTNG_VIEWER_H */
index 730eaf203102886b206aa5d267afc2f8bc9e98b6..02f676df8bc0e1bf56542e2c2394a5cabb59185c 100644 (file)
@@ -62,6 +62,7 @@
 #include "live.h"
 #include "health-relayd.h"
 #include "testpoint.h"
+#include "viewer-stream.h"
 
 /* command line options */
 char *opt_output_path;
@@ -1034,7 +1035,7 @@ static void destroy_stream(struct relay_stream *stream)
                }
        }
 
-       vstream = live_find_viewer_stream_by_id(stream->stream_handle);
+       vstream = viewer_stream_find_by_id(stream->stream_handle);
        if (vstream) {
                /*
                 * Set the last good value into the viewer stream. This is done
@@ -2379,7 +2380,7 @@ int relay_process_data(struct relay_command *cmd)
                                (stream->oldest_tracefile_id + 1) %
                                stream->tracefile_count;
                }
-               vstream = live_find_viewer_stream_by_id(stream->stream_handle);
+               vstream = viewer_stream_find_by_id(stream->stream_handle);
                if (vstream) {
                        /*
                         * The viewer is reading a file about to be
diff --git a/src/bin/lttng-relayd/session.c b/src/bin/lttng-relayd/session.c
new file mode 100644 (file)
index 0000000..07f1d9f
--- /dev/null
@@ -0,0 +1,45 @@
+/*
+ * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
+ *                      David Goulet <dgoulet@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.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+ * more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * this program; if not, write to the Free Software Foundation, Inc., 51
+ * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+ */
+
+#include "session.h"
+
+/*
+ * Lookup a session within the given hash table and session id. RCU read side
+ * lock MUST be acquired before calling this and as long as the caller has a
+ * reference to the object.
+ *
+ * Return session or NULL if not found.
+ */
+struct relay_session *session_find_by_id(struct lttng_ht *ht, uint64_t id)
+{
+       struct relay_session *session = NULL;
+       struct lttng_ht_node_ulong *node;
+       struct lttng_ht_iter iter;
+
+       assert(ht);
+
+       lttng_ht_lookup(ht, (void *)((unsigned long) id), &iter);
+       node = lttng_ht_iter_get_node_ulong(&iter);
+       if (!node) {
+               goto end;
+       }
+       session = caa_container_of(node, struct relay_session, session_n);
+
+end:
+       return session;
+}
diff --git a/src/bin/lttng-relayd/session.h b/src/bin/lttng-relayd/session.h
new file mode 100644 (file)
index 0000000..698e8ef
--- /dev/null
@@ -0,0 +1,72 @@
+/*
+ * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
+ *                      David Goulet <dgoulet@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.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+ * more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * this program; if not, write to the Free Software Foundation, Inc., 51
+ * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+ */
+
+#ifndef _SESSION_H
+#define _SESSION_H
+
+#include <limits.h>
+#include <inttypes.h>
+#include <pthread.h>
+
+#include <common/hashtable/hashtable.h>
+
+/*
+ * Represents a session for the relay point of view
+ */
+struct relay_session {
+       /*
+        * This session id is used to identify a set of stream to a tracing session
+        * but also make sure we have a unique session id associated with a session
+        * daemon which can provide multiple data source.
+        */
+       uint64_t id;
+       struct lttcomm_sock *sock;
+       char session_name[NAME_MAX];
+       char hostname[HOST_NAME_MAX];
+       uint32_t live_timer;
+       struct lttng_ht_node_ulong session_n;
+       struct rcu_head rcu_node;
+       uint32_t viewer_attached;
+       uint32_t stream_count;
+       /* Tell if this session is for a snapshot or not. */
+       unsigned int snapshot:1;
+
+       /*
+        * Indicate version protocol for this session. This is especially useful
+        * for the data thread that has no idea which version it operates on since
+        * linking control/data sockets is non trivial.
+        */
+       uint64_t minor;
+       uint64_t major;
+       /*
+        * Flag checked and exchanged with uatomic_cmpxchg to tell the
+        * viewer-side if new streams got added since the last check.
+        */
+       unsigned long new_streams;
+
+       /*
+        * Used to synchronize the process where we flag every streams readiness
+        * for the viewer when the streams_sent message is received and the viewer
+        * process of sending those streams.
+        */
+       pthread_mutex_t viewer_ready_lock;
+};
+
+struct relay_session *session_find_by_id(struct lttng_ht *ht, uint64_t id);
+
+#endif /* _SESSION_H */
diff --git a/src/bin/lttng-relayd/viewer-stream.c b/src/bin/lttng-relayd/viewer-stream.c
new file mode 100644 (file)
index 0000000..3953d7d
--- /dev/null
@@ -0,0 +1,297 @@
+/*
+ * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
+ *                      David Goulet <dgoulet@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.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+ * more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * this program; if not, write to the Free Software Foundation, Inc., 51
+ * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+ */
+
+#define _GNU_SOURCE
+#include <common/common.h>
+#include <common/index/index.h>
+
+#include "lttng-relayd.h"
+#include "viewer-stream.h"
+
+static void free_stream(struct relay_viewer_stream *stream)
+{
+       assert(stream);
+
+       free(stream->path_name);
+       free(stream->channel_name);
+       free(stream);
+}
+
+static void deferred_free_viewer_stream(struct rcu_head *head)
+{
+       struct relay_viewer_stream *stream =
+               caa_container_of(head, struct relay_viewer_stream, rcu_node);
+
+       free_stream(stream);
+}
+
+struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream,
+               enum lttng_viewer_seek seek_t)
+{
+       struct relay_viewer_stream *vstream;
+
+       assert(stream);
+
+       vstream = zmalloc(sizeof(*vstream));
+       if (!vstream) {
+               PERROR("relay viewer stream zmalloc");
+               goto error;
+       }
+
+       vstream->session_id = stream->session->id;
+       vstream->stream_handle = stream->stream_handle;
+       vstream->path_name = strndup(stream->path_name, LTTNG_VIEWER_PATH_MAX);
+       vstream->channel_name = strndup(stream->channel_name,
+                       LTTNG_VIEWER_NAME_MAX);
+       vstream->tracefile_count = stream->tracefile_count;
+       vstream->metadata_flag = stream->metadata_flag;
+       vstream->tracefile_count_last = -1ULL;
+
+       switch (seek_t) {
+       case VIEWER_SEEK_BEGINNING:
+               vstream->tracefile_count_current = stream->oldest_tracefile_id;
+               break;
+       case VIEWER_SEEK_LAST:
+               vstream->tracefile_count_current = stream->tracefile_count_current;
+               break;
+       default:
+               assert(0);
+               goto error;
+       }
+
+       vstream->ctf_trace = stream->ctf_trace;
+       if (vstream->metadata_flag) {
+               vstream->ctf_trace->viewer_metadata_stream = vstream;
+       }
+       uatomic_inc(&vstream->ctf_trace->refcount);
+
+       /* Globally visible after the add unique. */
+       lttng_ht_node_init_u64(&vstream->stream_n, stream->stream_handle);
+       lttng_ht_add_unique_u64(viewer_streams_ht, &vstream->stream_n);
+
+       vstream->index_read_fd = -1;
+       vstream->read_fd = -1;
+
+       /*
+        * This is to avoid a race between the initialization of this object and
+        * the close of the given stream. If the stream is unable to find this
+        * viewer stream when closing, this copy will at least take the latest
+        * value. We also need that for the seek_last.
+        */
+       vstream->total_index_received = stream->total_index_received;
+
+       /*
+        * If we never received an index for the current stream, delay the opening
+        * of the index, otherwise open it right now.
+        */
+       if (vstream->tracefile_count_current == stream->tracefile_count_current
+                       && vstream->total_index_received == 0) {
+               vstream->index_read_fd = -1;
+       } else {
+               int read_fd;
+
+               read_fd = index_open(vstream->path_name, vstream->channel_name,
+                               vstream->tracefile_count, vstream->tracefile_count_current);
+               if (read_fd < 0) {
+                       goto error;
+               }
+               vstream->index_read_fd = read_fd;
+       }
+
+       if (seek_t == VIEWER_SEEK_LAST && vstream->index_read_fd >= 0) {
+               off_t lseek_ret;
+
+               lseek_ret = lseek(vstream->index_read_fd,
+                               vstream->total_index_received * sizeof(struct ctf_packet_index),
+                               SEEK_CUR);
+               if (lseek_ret < 0) {
+                       goto error;
+               }
+               vstream->last_sent_index = vstream->total_index_received;
+       }
+
+       return vstream;
+
+error:
+       if (vstream) {
+               free_stream(vstream);
+       }
+       return NULL;
+}
+
+void viewer_stream_delete(struct relay_viewer_stream *stream)
+{
+       int ret;
+       struct lttng_ht_iter iter;
+
+       iter.iter.node = &stream->stream_n.node;
+       ret = lttng_ht_del(viewer_streams_ht, &iter);
+       assert(!ret);
+}
+
+void viewer_stream_destroy(struct relay_viewer_stream *stream)
+{
+       int ret;
+       unsigned long ret_ref;
+
+       assert(stream);
+       ret_ref = uatomic_add_return(&stream->ctf_trace->refcount, -1);
+       assert(ret_ref >= 0);
+
+       if (stream->read_fd >= 0) {
+               ret = close(stream->read_fd);
+               if (ret < 0) {
+                       PERROR("close read_fd");
+               }
+       }
+       if (stream->index_read_fd >= 0) {
+               ret = close(stream->index_read_fd);
+               if (ret < 0) {
+                       PERROR("close index_read_fd");
+               }
+       }
+
+       /*
+        * If the only stream left in the HT is the metadata stream,
+        * we need to remove it because we won't detect a EOF for this
+        * stream.
+        */
+       if (ret_ref == 1 && stream->ctf_trace->viewer_metadata_stream) {
+               viewer_stream_delete(stream->ctf_trace->viewer_metadata_stream);
+               viewer_stream_destroy(stream->ctf_trace->viewer_metadata_stream);
+               stream->ctf_trace->metadata_stream = NULL;
+               DBG("Freeing ctf_trace %" PRIu64, stream->ctf_trace->id);
+               /*
+                * The streaming-side is already closed and we can't receive a new
+                * stream concurrently at this point (since the session is being
+                * destroyed), so when we detect the refcount equals 0, we are the
+                * only owners of the ctf_trace and we can free it ourself.
+                */
+               free(stream->ctf_trace);
+       }
+
+       call_rcu(&stream->rcu_node, deferred_free_viewer_stream);
+}
+
+/*
+ * Find viewer stream by id. RCU read side lock MUST be acquired.
+ *
+ * Return stream if found else NULL.
+ */
+struct relay_viewer_stream *viewer_stream_find_by_id(uint64_t id)
+{
+       struct lttng_ht_node_u64 *node;
+       struct lttng_ht_iter iter;
+       struct relay_viewer_stream *stream = NULL;
+
+       lttng_ht_lookup(viewer_streams_ht, &id, &iter);
+       node = lttng_ht_iter_get_node_u64(&iter);
+       if (!node) {
+               DBG("Relay viewer stream %" PRIu64 " not found", id);
+               goto end;
+       }
+       stream = caa_container_of(node, struct relay_viewer_stream, stream_n);
+
+end:
+       return stream;
+}
+
+/*
+ * Rotate a stream to the next tracefile.
+ *
+ * Returns 0 on success, 1 on EOF, a negative value on error.
+ */
+int viewer_stream_rotate(struct relay_viewer_stream *vstream,
+               struct relay_stream *stream)
+{
+       int ret;
+       uint64_t tracefile_id;
+
+       assert(vstream);
+
+       tracefile_id = (vstream->tracefile_count_current + 1) %
+               vstream->tracefile_count;
+
+       /* Detect the last tracefile to open. */
+       if (vstream->tracefile_count_last != -1ULL &&
+                       vstream->tracefile_count_last ==
+                       vstream->tracefile_count_current) {
+               ret = 1;
+               goto end;
+       }
+
+       /*
+        * If the stream on the streaming side still exists, lock to execute
+        * rotation in order to avoid races between a modification on the index
+        * values.
+        */
+       if (stream) {
+               pthread_mutex_lock(&stream->viewer_stream_rotation_lock);
+       }
+
+       /*
+        * The writer and the reader are not working in the same tracefile, we can
+        * read up to EOF, we don't care about the total_index_received.
+        */
+       if (!stream || (stream->tracefile_count_current != tracefile_id)) {
+               vstream->close_write_flag = 1;
+       } else {
+               /*
+                * We are opening a file that is still open in write, make sure we
+                * limit our reading to the number of indexes received.
+                */
+               vstream->close_write_flag = 0;
+               if (stream) {
+                       vstream->total_index_received = stream->total_index_received;
+               }
+       }
+       vstream->tracefile_count_current = tracefile_id;
+
+       ret = close(vstream->index_read_fd);
+       if (ret < 0) {
+               PERROR("close index file %d", vstream->index_read_fd);
+       }
+       vstream->index_read_fd = -1;
+
+       ret = close(vstream->read_fd);
+       if (ret < 0) {
+               PERROR("close tracefile %d", vstream->read_fd);
+       }
+       vstream->read_fd = -1;
+
+       pthread_mutex_lock(&vstream->overwrite_lock);
+       vstream->abort_flag = 0;
+       pthread_mutex_unlock(&vstream->overwrite_lock);
+
+       if (stream) {
+               pthread_mutex_unlock(&stream->viewer_stream_rotation_lock);
+       }
+
+       ret = index_open(vstream->path_name, vstream->channel_name,
+                       vstream->tracefile_count, vstream->tracefile_count_current);
+       if (ret < 0) {
+               goto error;
+       }
+       vstream->index_read_fd = ret;
+
+       ret = 0;
+
+end:
+error:
+       return ret;
+}
diff --git a/src/bin/lttng-relayd/viewer-stream.h b/src/bin/lttng-relayd/viewer-stream.h
new file mode 100644 (file)
index 0000000..793fa7c
--- /dev/null
@@ -0,0 +1,87 @@
+/*
+ * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
+ *                      David Goulet <dgoulet@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.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+ * more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * this program; if not, write to the Free Software Foundation, Inc., 51
+ * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+ */
+
+#ifndef _VIEWER_STREAM_H
+#define _VIEWER_STREAM_H
+
+#include <limits.h>
+#include <inttypes.h>
+#include <pthread.h>
+
+#include <common/hashtable/hashtable.h>
+
+#include "ctf-trace.h"
+#include "lttng-viewer-abi.h"
+
+/* Stub */
+struct relay_stream;
+
+/*
+ * Shadow copy of the relay_stream structure for the viewer side.  The only
+ * fields updated by the writer (streaming side) after allocation are :
+ * total_index_received and close_flag. Everything else is updated by the
+ * reader (viewer side).
+ */
+struct relay_viewer_stream {
+       uint64_t stream_handle;
+       uint64_t session_id;
+       int read_fd;
+       int index_read_fd;
+       char *path_name;
+       char *channel_name;
+       uint64_t last_sent_index;
+       uint64_t total_index_received;
+       uint64_t tracefile_count;
+       uint64_t tracefile_count_current;
+       /* Stop after reading this tracefile. */
+       uint64_t tracefile_count_last;
+       struct lttng_ht_node_u64 stream_n;
+       struct rcu_head rcu_node;
+       struct ctf_trace *ctf_trace;
+       /*
+        * This lock blocks only when the writer is about to start overwriting
+        * a file currently read by the reader.
+        *
+        * This is nested INSIDE the viewer_stream_rotation_lock.
+        */
+       pthread_mutex_t overwrite_lock;
+       /* Information telling us if the stream is a metadata stream. */
+       unsigned int metadata_flag:1;
+       /*
+        * Information telling us that the stream is closed in write, so
+        * we don't expect new indexes and we can read up to EOF.
+        */
+       unsigned int close_write_flag:1;
+       /*
+        * If the streaming side closes a FD in use in the viewer side,
+        * it sets this flag to inform that it is a normal error.
+        */
+       unsigned int abort_flag:1;
+       /* Indicates if this stream has been sent to a viewer client. */
+       unsigned int sent_flag:1;
+};
+
+struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream,
+               enum lttng_viewer_seek seek_t);
+struct relay_viewer_stream *viewer_stream_find_by_id(uint64_t id);
+void viewer_stream_destroy(struct relay_viewer_stream *stream);
+void viewer_stream_delete(struct relay_viewer_stream *stream);
+int viewer_stream_rotate(struct relay_viewer_stream *vstream,
+               struct relay_stream *stream);
+
+#endif /* _VIEWER_STREAM_H */
index 2946f7bab3335e5456d9628d9e0a3c6707e0eca6..54689677ee4a2f3dd3c56a6f06236115f659ead6 100644 (file)
@@ -19,6 +19,8 @@
 #define _GNU_SOURCE
 #include <assert.h>
 #include <sys/stat.h>
+#include <sys/types.h>
+#include <fcntl.h>
 
 #include <common/common.h>
 #include <common/defaults.h>
@@ -108,3 +110,76 @@ ssize_t index_write(int fd, struct ctf_packet_index *index, size_t len)
 
        return ret;
 }
+
+/*
+ * Open index file using a given path, channel name and tracefile count.
+ *
+ * Return read only FD on success or else a negative value.
+ */
+int index_open(const char *path_name, const char *channel_name,
+               uint64_t tracefile_count, uint64_t tracefile_count_current)
+{
+       int ret, read_fd;
+       ssize_t read_len;
+       char fullpath[PATH_MAX];
+       struct ctf_packet_index_file_hdr hdr;
+
+       assert(path_name);
+       assert(channel_name);
+
+       if (tracefile_count > 0) {
+               ret = snprintf(fullpath, sizeof(fullpath), "%s/" DEFAULT_INDEX_DIR "/%s_%"
+                               PRIu64 DEFAULT_INDEX_FILE_SUFFIX, path_name,
+                               channel_name, tracefile_count_current);
+       } else {
+               ret = snprintf(fullpath, sizeof(fullpath), "%s/" DEFAULT_INDEX_DIR "/%s"
+                               DEFAULT_INDEX_FILE_SUFFIX, path_name, channel_name);
+       }
+       if (ret < 0) {
+               PERROR("snprintf index path");
+               goto error;
+       }
+
+       DBG("Index opening file %s in read only", fullpath);
+       read_fd = open(fullpath, O_RDONLY);
+       if (read_fd < 0) {
+               if (errno == ENOENT) {
+                       ret = -ENOENT;
+               } else {
+                       PERROR("opening index in read-only");
+               }
+               goto error;
+       }
+
+       read_len = lttng_read(read_fd, &hdr, sizeof(hdr));
+       if (read_len < 0) {
+               PERROR("Reading index header");
+               goto error_close;
+       }
+
+       if (be32toh(hdr.magic) != CTF_INDEX_MAGIC) {
+               ERR("Invalid header magic");
+               goto error_close;
+       }
+       if (be32toh(hdr.index_major) != CTF_INDEX_MAJOR ||
+                       be32toh(hdr.index_minor) != CTF_INDEX_MINOR) {
+               ERR("Invalid header version");
+               goto error_close;
+       }
+
+       return read_fd;
+
+error_close:
+       if (read_fd >= 0) {
+               int close_ret;
+
+               close_ret = close(read_fd);
+               if (close_ret < 0) {
+                       PERROR("close read fd %d", read_fd);
+               }
+       }
+       ret = -1;
+
+error:
+       return ret;
+}
index 744fe0107f82e22a1f73c4fb208beec964012dab..a7e6aee07a046a2b7b097f76f0a920dcf2bc1ac8 100644 (file)
@@ -26,5 +26,7 @@
 int index_create_file(char *path_name, char *stream_name, int uid, int gid,
                uint64_t size, uint64_t count);
 ssize_t index_write(int fd, struct ctf_packet_index *index, size_t len);
+int index_open(const char *path_name, const char *channel_name,
+               uint64_t tracefile_count, uint64_t tracefile_count_current);
 
 #endif /* _INDEX_H */
index 75d4e4fdbd9a3f5b41a7f06186dfa2d0b26dabb2..e597890f4a0bcd01153bda8b6b8bd8b80b97623c 100644 (file)
@@ -40,7 +40,7 @@
 #include <bin/lttng-sessiond/session.h>
 #include <common/common.h>
 
-#include <bin/lttng-relayd/lttng-viewer.h>
+#include <bin/lttng-relayd/lttng-viewer-abi.h>
 #include <common/index/ctf-index.h>
 
 #define SESSION1 "test1"
This page took 0.094023 seconds and 4 git commands to generate.