Fix: relayd: live connection fails to open file during clear

author Jérémie Galarneau <jeremie.galarneau@efficios.com>

Tue, 28 Jul 2020 15:29:18 +0000 (11:29 -0400)

committer Jérémie Galarneau <jeremie.galarneau@efficios.com>

Wed, 29 Jul 2020 16:38:48 +0000 (12:38 -0400)
author Jérémie Galarneau <jeremie.galarneau@efficios.com>
Tue, 28 Jul 2020 15:29:18 +0000 (11:29 -0400)
committer Jérémie Galarneau <jeremie.galarneau@efficios.com>
Wed, 29 Jul 2020 16:38:48 +0000 (12:38 -0400)
diff --git a/src/bin/lttng-relayd/live.c b/src/bin/lttng-relayd/live.c

index d5f04582419dffec1466c60c4de449cf34948a03..7d6dc1bc20a03a5af6cafe1e5ef170be1b7f5cb8 100644 (file)
--- a/src/bin/lttng-relayd/live.c
+++ b/src/bin/lttng-relayd/live.c
@@ -268,8 +268,8 @@ end_unlock:
   *
   * Return 0 on success or else a negative value.
   */
   *
   * Return 0 on success or else a negative value.
   */
-static int make_viewer_streams(struct relay_session *session,
-               struct lttng_trace_chunk *viewer_trace_chunk,
+static int make_viewer_streams(struct relay_session *relay_session,
+               struct relay_viewer_session *viewer_session,
                 enum lttng_viewer_seek seek_t,
                 uint32_t *nb_total,
                 uint32_t *nb_unsent,
                 enum lttng_viewer_seek seek_t,
                 uint32_t *nb_total,
                 uint32_t *nb_unsent,
@@ -279,18 +279,19 @@ static int make_viewer_streams(struct relay_session *session,
         int ret;
         struct lttng_ht_iter iter;
         struct ctf_trace *ctf_trace;
         int ret;
         struct lttng_ht_iter iter;
         struct ctf_trace *ctf_trace;
+       struct relay_stream *relay_stream = NULL;
  
  
-       assert(session);
-       ASSERT_LOCKED(session->lock);
+       assert(relay_session);
+       ASSERT_LOCKED(relay_session->lock);
  
  
-       if (!viewer_trace_chunk) {
+       if (!viewer_session->current_trace_chunk) {
                 ERR("Internal error: viewer session associated with session \"%s\" has a NULL trace chunk",
                 ERR("Internal error: viewer session associated with session \"%s\" has a NULL trace chunk",
-                               session->session_name);
+                               relay_session->session_name);
                 ret = -1;
                 goto error;
         }
  
                 ret = -1;
                 goto error;
         }
  
-       if (session->connection_closed) {
+       if (relay_session->connection_closed) {
                 *closed = true;
         }
  
                 *closed = true;
         }
  
@@ -299,10 +300,9 @@ static int make_viewer_streams(struct relay_session *session,
          * used for a the given session id only.
          */
         rcu_read_lock();
          * used for a the given session id only.
          */
         rcu_read_lock();
-       cds_lfht_for_each_entry(session->ctf_traces_ht->ht, &iter.iter, ctf_trace,
-                       node.node) {
+       cds_lfht_for_each_entry (relay_session->ctf_traces_ht->ht, &iter.iter,
+                       ctf_trace, node.node) {
                 bool trace_has_metadata_stream = false;
                 bool trace_has_metadata_stream = false;
-               struct relay_stream *stream;
  
                 health_code_update();
  
  
                 health_code_update();
  
@@ -314,15 +314,23 @@ static int make_viewer_streams(struct relay_session *session,
                  * Iterate over all the streams of the trace to see if we have a
                  * metadata stream.
                  */
                  * Iterate over all the streams of the trace to see if we have a
                  * metadata stream.
                  */
-               cds_list_for_each_entry_rcu(
-                               stream, &ctf_trace->stream_list, stream_node)
+               cds_list_for_each_entry_rcu(relay_stream,
+                               &ctf_trace->stream_list, stream_node)
                 {
                 {
-                       if (stream->is_metadata) {
+                       bool is_metadata_stream;
+
+                       pthread_mutex_lock(&relay_stream->lock);
+                       is_metadata_stream = relay_stream->is_metadata;
+                       pthread_mutex_unlock(&relay_stream->lock);
+
+                       if (is_metadata_stream) {
                                 trace_has_metadata_stream = true;
                                 break;
                         }
                 }
  
                                 trace_has_metadata_stream = true;
                                 break;
                         }
                 }
  
+               relay_stream = NULL;
+
                 /*
                  * If there is no metadata stream in this trace at the moment
                  * and we never sent one to the viewer, skip the trace. We
                 /*
                  * If there is no metadata stream in this trace at the moment
                  * and we never sent one to the viewer, skip the trace. We
@@ -334,35 +342,72 @@ static int make_viewer_streams(struct relay_session *session,
                         continue;
                 }
  
                         continue;
                 }
  
-               cds_list_for_each_entry_rcu(stream, &ctf_trace->stream_list, stream_node) {
-                       struct relay_viewer_stream *vstream;
+               cds_list_for_each_entry_rcu(relay_stream,
+                               &ctf_trace->stream_list, stream_node)
+               {
+                       struct relay_viewer_stream *viewer_stream;
  
  
-                       if (!stream_get(stream)) {
+                       if (!stream_get(relay_stream)) {
                                 continue;
                         }
                                 continue;
                         }
+
+                       pthread_mutex_lock(&relay_stream->lock);
                         /*
                          * stream published is protected by the session lock.
                          */
                         /*
                          * stream published is protected by the session lock.
                          */
-                       if (!stream->published) {
+                       if (!relay_stream->published) {
                                 goto next;
                         }
                                 goto next;
                         }
-                       vstream = viewer_stream_get_by_id(stream->stream_handle);
-                       if (!vstream) {
+                       viewer_stream = viewer_stream_get_by_id(
+                                       relay_stream->stream_handle);
+                       if (!viewer_stream) {
+                               struct lttng_trace_chunk *viewer_stream_trace_chunk;
+
                                 /*
                                  * Save that we sent the metadata stream to the
                                  * viewer. So that we know what trace the viewer
                                  * is aware of.
                                  */
                                 /*
                                  * Save that we sent the metadata stream to the
                                  * viewer. So that we know what trace the viewer
                                  * is aware of.
                                  */
-                               if (stream->is_metadata) {
-                                       ctf_trace->metadata_stream_sent_to_viewer =
-                                                       true;
+                               if (relay_stream->is_metadata) {
+                                       ctf_trace->metadata_stream_sent_to_viewer = true;
                                 }
                                 }
-                               vstream = viewer_stream_create(stream,
-                                               viewer_trace_chunk, seek_t);
-                               if (!vstream) {
+
+                               /*
+                                * If a rotation is ongoing, use a copy of the
+                                * relay stream's chunk to ensure the stream
+                                * files exist.
+                                *
+                                * Otherwise, the viewer session's current trace
+                                * chunk can be used safely.
+                                */
+                               if ((relay_stream->ongoing_rotation.is_set ||
+                                                   relay_session->ongoing_rotation) &&
+                                               relay_stream->trace_chunk) {
+                                       viewer_stream_trace_chunk = lttng_trace_chunk_copy(
+                                                       relay_stream->trace_chunk);
+                                       if (!viewer_stream_trace_chunk) {
+                                               ret = -1;
+                                               ctf_trace_put(ctf_trace);
+                                               goto error_unlock;
+                                       }
+                               } else {
+                                       const bool reference_acquired = lttng_trace_chunk_get(
+                                                       viewer_session->current_trace_chunk);
+
+                                       assert(reference_acquired);
+                                       viewer_stream_trace_chunk =
+                                                       viewer_session->current_trace_chunk;
+                               }
+
+                               viewer_stream = viewer_stream_create(
+                                               relay_stream,
+                                               viewer_stream_trace_chunk,
+                                               seek_t);
+                               lttng_trace_chunk_put(viewer_stream_trace_chunk);
+                               viewer_stream_trace_chunk = NULL;
+                               if (!viewer_stream) {
                                         ret = -1;
                                         ctf_trace_put(ctf_trace);
                                         ret = -1;
                                         ctf_trace_put(ctf_trace);
-                                       stream_put(stream);
                                         goto error_unlock;
                                 }
  
                                         goto error_unlock;
                                 }
  
@@ -374,36 +419,40 @@ static int make_viewer_streams(struct relay_session *session,
                                  * Ensure a self-reference is preserved even
                                  * after we have put our local reference.
                                  */
                                  * Ensure a self-reference is preserved even
                                  * after we have put our local reference.
                                  */
-                               if (!viewer_stream_get(vstream)) {
+                               if (!viewer_stream_get(viewer_stream)) {
                                         ERR("Unable to get self-reference on viewer stream, logic error.");
                                         abort();
                                 }
                         } else {
                                         ERR("Unable to get self-reference on viewer stream, logic error.");
                                         abort();
                                 }
                         } else {
-                               if (!vstream->sent_flag && nb_unsent) {
+                               if (!viewer_stream->sent_flag && nb_unsent) {
                                         /* Update number of unsent stream counter. */
                                         (*nb_unsent)++;
                                 }
                         }
                         /* Update number of total stream counter. */
                         if (nb_total) {
                                         /* Update number of unsent stream counter. */
                                         (*nb_unsent)++;
                                 }
                         }
                         /* Update number of total stream counter. */
                         if (nb_total) {
-                               if (stream->is_metadata) {
-                                       if (!stream->closed ||
-                                                       stream->metadata_received > vstream->metadata_sent) {
+                               if (relay_stream->is_metadata) {
+                                       if (!relay_stream->closed ||
+                                                       relay_stream->metadata_received >
+                                                                       viewer_stream->metadata_sent) {
                                                 (*nb_total)++;
                                         }
                                 } else {
                                                 (*nb_total)++;
                                         }
                                 } else {
-                                       if (!stream->closed ||
-                                               !(((int64_t) (stream->prev_data_seq - stream->last_net_seq_num)) >= 0)) {
-
+                                       if (!relay_stream->closed ||
+                                                       !(((int64_t)(relay_stream->prev_data_seq -
+                                                                         relay_stream->last_net_seq_num)) >=
+                                                                       0)) {
                                                 (*nb_total)++;
                                         }
                                 }
                         }
                         /* Put local reference. */
                                                 (*nb_total)++;
                                         }
                                 }
                         }
                         /* Put local reference. */
-                       viewer_stream_put(vstream);
+                       viewer_stream_put(viewer_stream);
                 next:
                 next:
-                       stream_put(stream);
+                       pthread_mutex_unlock(&relay_stream->lock);
+                       stream_put(relay_stream);
                 }
                 }
+               relay_stream = NULL;
                 ctf_trace_put(ctf_trace);
         }
  
                 ctf_trace_put(ctf_trace);
         }
  
@@ -412,6 +461,11 @@ static int make_viewer_streams(struct relay_session *session,
  error_unlock:
         rcu_read_unlock();
  error:
  error_unlock:
         rcu_read_unlock();
  error:
+       if (relay_stream) {
+               pthread_mutex_unlock(&relay_stream->lock);
+               stream_put(relay_stream);
+       }
+
         return ret;
  }
  
         return ret;
  }
  
@@ -1088,7 +1142,7 @@ int viewer_get_new_streams(struct relay_connection *conn)
  
         pthread_mutex_lock(&session->lock);
         ret = make_viewer_streams(session,
  
         pthread_mutex_lock(&session->lock);
         ret = make_viewer_streams(session,
-                       conn->viewer_session->current_trace_chunk,
+                       conn->viewer_session,
                         LTTNG_VIEWER_SEEK_LAST, &nb_total, &nb_unsent,
                         &nb_created, &closed);
         if (ret < 0) {
                         LTTNG_VIEWER_SEEK_LAST, &nb_total, &nb_unsent,
                         &nb_created, &closed);
         if (ret < 0) {
@@ -1237,7 +1291,7 @@ int viewer_attach_session(struct relay_connection *conn)
         }
  
         ret = make_viewer_streams(session,
         }
  
         ret = make_viewer_streams(session,
-                       conn->viewer_session->current_trace_chunk, seek_type,
+                       conn->viewer_session, seek_type,
                         &nb_streams, NULL, NULL, &closed);
         if (ret < 0) {
                 goto end_put_session;
                         &nb_streams, NULL, NULL, &closed);
         if (ret < 0) {
                 goto end_put_session;
diff --git a/src/bin/lttng-relayd/viewer-session.h b/src/bin/lttng-relayd/viewer-session.h

index ebde2dd1dbfc77cfb08c541f898ed89f76253138..2895db0ef43230ed4d40b5dad4a24da434abf22f 100644 (file)
--- a/src/bin/lttng-relayd/viewer-session.h
+++ b/src/bin/lttng-relayd/viewer-session.h
@@ -30,7 +30,6 @@ struct relay_viewer_session {
          */
         struct cds_list_head session_list;      /* RCU list. */
         pthread_mutex_t session_list_lock;      /* Protects list updates. */
          */
         struct cds_list_head session_list;      /* RCU list. */
         pthread_mutex_t session_list_lock;      /* Protects list updates. */
-       /* Once set, the current trace chunk of a viewer must not change. */
         struct lttng_trace_chunk *current_trace_chunk;
  };
  
         struct lttng_trace_chunk *current_trace_chunk;
  };
  
diff --git a/src/bin/lttng-relayd/viewer-stream.c b/src/bin/lttng-relayd/viewer-stream.c

index 05499215a8049e5e98e239b7e8f50d030230328e..f07e062f8d303a6b1aeeb0520a263009a05eae3d 100644 (file)
--- a/src/bin/lttng-relayd/viewer-stream.c
+++ b/src/bin/lttng-relayd/viewer-stream.c
@@ -34,14 +34,15 @@ static void viewer_stream_destroy_rcu(struct rcu_head *head)
         viewer_stream_destroy(vstream);
  }
  
         viewer_stream_destroy(vstream);
  }
  
+/* Relay stream's lock must be held by the caller. */
  struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream,
  struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream,
-               struct lttng_trace_chunk *viewer_trace_chunk,
+               struct lttng_trace_chunk *trace_chunk,
                 enum lttng_viewer_seek seek_t)
  {
         struct relay_viewer_stream *vstream = NULL;
                 enum lttng_viewer_seek seek_t)
  {
         struct relay_viewer_stream *vstream = NULL;
-       const bool acquired_reference = lttng_trace_chunk_get(
-                       viewer_trace_chunk);
+       const bool acquired_reference = lttng_trace_chunk_get(trace_chunk);
  
  
+       ASSERT_LOCKED(stream->lock);
         if (!acquired_reference) {
                 goto error;
         }
         if (!acquired_reference) {
                 goto error;
         }
@@ -52,8 +53,8 @@ struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream,
                 goto error;
         }
  
                 goto error;
         }
  
-       vstream->stream_file.trace_chunk = viewer_trace_chunk;
-       viewer_trace_chunk = NULL;
+       vstream->stream_file.trace_chunk = trace_chunk;
+       trace_chunk = NULL;
         vstream->path_name = lttng_strndup(stream->path_name, LTTNG_VIEWER_PATH_MAX);
         if (vstream->path_name == NULL) {
                 PERROR("relay viewer path_name alloc");
         vstream->path_name = lttng_strndup(stream->path_name, LTTNG_VIEWER_PATH_MAX);
         if (vstream->path_name == NULL) {
                 PERROR("relay viewer path_name alloc");
@@ -72,8 +73,6 @@ struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream,
         }
         vstream->stream = stream;
  
         }
         vstream->stream = stream;
  
-       pthread_mutex_lock(&stream->lock);
-
         if (stream->is_metadata && stream->trace->viewer_metadata_stream) {
                 ERR("Cannot attach viewer metadata stream to trace (busy).");
                 goto error_unlock;
         if (stream->is_metadata && stream->trace->viewer_metadata_stream) {
                 ERR("Cannot attach viewer metadata stream to trace (busy).");
                 goto error_unlock;
@@ -183,7 +182,6 @@ struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream,
                 rcu_assign_pointer(stream->trace->viewer_metadata_stream,
                                 vstream);
         }
                 rcu_assign_pointer(stream->trace->viewer_metadata_stream,
                                 vstream);
         }
-       pthread_mutex_unlock(&stream->lock);
  
         /* Globally visible after the add unique. */
         lttng_ht_node_init_u64(&vstream->stream_n, stream->stream_handle);
  
         /* Globally visible after the add unique. */
         lttng_ht_node_init_u64(&vstream->stream_n, stream->stream_handle);
@@ -198,8 +196,8 @@ error:
         if (vstream) {
                 viewer_stream_destroy(vstream);
         }
         if (vstream) {
                 viewer_stream_destroy(vstream);
         }
-       if (viewer_trace_chunk && acquired_reference) {
-               lttng_trace_chunk_put(viewer_trace_chunk);
+       if (trace_chunk && acquired_reference) {
+               lttng_trace_chunk_put(trace_chunk);
         }
         return NULL;
  }
         }
         return NULL;
  }
author	Jérémie Galarneau <jeremie.galarneau@efficios.com>
	Tue, 28 Jul 2020 15:29:18 +0000 (11:29 -0400)
committer	Jérémie Galarneau <jeremie.galarneau@efficios.com>
	Wed, 29 Jul 2020 16:38:48 +0000 (12:38 -0400)
src/bin/lttng-relayd/live.c		patch \| blob \| blame \| history
src/bin/lttng-relayd/viewer-session.h		patch \| blob \| blame \| history
src/bin/lttng-relayd/viewer-stream.c		patch \| blob \| blame \| history