X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Flive.c;h=06c1c15134cf63af055729b6b28f75d0489962ce;hp=5b9e4248c0e779deadc7b604ace4d928f7c9634f;hb=b5a6470f372799b28d3d20603c1c0c8e5871dd63;hpb=6763619c3f295a1502e77ca9563d7a21d68b25c7 diff --git a/src/bin/lttng-relayd/live.c b/src/bin/lttng-relayd/live.c index 5b9e4248c..06c1c1513 100644 --- a/src/bin/lttng-relayd/live.c +++ b/src/bin/lttng-relayd/live.c @@ -153,32 +153,33 @@ ssize_t send_response(struct lttcomm_sock *sock, void *buf, size_t size) } /* - * Atomically check if new streams got added in the session since the last - * check and reset the flag to 0. + * Atomically check if new streams got added in one of the sessions attached + * 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 check_new_streams(struct relay_connection *conn) { - int ret; - unsigned long current_val; struct relay_session *session; + unsigned long current_val; + int ret = 0; - 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; + if (!conn->viewer_session) { + goto end; + } + cds_list_for_each_entry(session, + &conn->viewer_session->sessions_head, + viewer_session_list) { + current_val = uatomic_cmpxchg(&session->new_streams, 1, 0); + ret = current_val; + if (ret == 1) { + goto end; + } } - current_val = uatomic_cmpxchg(&session->new_streams, 1, 0); - ret = current_val; - -error: +end: return ret; } @@ -349,7 +350,7 @@ void stop_threads(void) /* Stopping all threads */ DBG("Terminating all live threads"); - ret = notify_thread_pipe(live_conn_pipe[1]); + ret = notify_thread_pipe(thread_quit_pipe[1]); if (ret < 0) { ERR("write error on thread quit pipe"); } @@ -378,7 +379,7 @@ int create_thread_poll_set(struct lttng_poll_event *events, int size) } /* Add quit pipe */ - ret = lttng_poll_add(events, live_conn_pipe[0], LPOLLIN | LPOLLERR); + ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN | LPOLLERR); if (ret < 0) { goto error; } @@ -395,9 +396,9 @@ error: * Return 1 if it was triggered else 0; */ static -int check_live_conn_pipe(int fd, uint32_t events) +int check_thread_quit_pipe(int fd, uint32_t events) { - if (fd == live_conn_pipe[0] && (events & LPOLLIN)) { + if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) { return 1; } @@ -514,7 +515,7 @@ restart: pollfd = LTTNG_POLL_GETFD(&events, i); /* Thread quit pipe has been closed. Killing thread. */ - ret = check_live_conn_pipe(pollfd, revents); + ret = check_thread_quit_pipe(pollfd, revents); if (ret) { err = 0; goto exit; @@ -693,6 +694,7 @@ int viewer_connect(struct relay_connection *conn) health_code_update(); + memset(&reply, 0, sizeof(reply)); reply.major = RELAYD_VERSION_COMM_MAJOR; reply.minor = RELAYD_VERSION_COMM_MINOR; @@ -835,6 +837,72 @@ end: return found; } +/* + * Delete all streams for a specific session ID. + */ +static void destroy_viewer_streams_by_session(struct relay_session *session) +{ + struct relay_viewer_stream *stream; + struct lttng_ht_iter iter; + + assert(session); + + rcu_read_lock(); + cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, stream, + stream_n.node) { + struct ctf_trace *ctf_trace; + + health_code_update(); + if (stream->session_id != session->id) { + continue; + } + + ctf_trace = ctf_trace_find_by_path(session->ctf_traces_ht, + stream->path_name); + assert(ctf_trace); + + viewer_stream_delete(stream); + + if (stream->metadata_flag) { + ctf_trace->metadata_sent = 0; + ctf_trace->viewer_metadata_stream = NULL; + } + + viewer_stream_destroy(ctf_trace, stream); + } + rcu_read_unlock(); +} + +static void try_destroy_streams(struct relay_session *session) +{ + struct ctf_trace *ctf_trace; + struct lttng_ht_iter iter; + + assert(session); + + cds_lfht_for_each_entry(session->ctf_traces_ht->ht, &iter.iter, ctf_trace, + node.node) { + /* Attempt to destroy the ctf trace of that session. */ + ctf_trace_try_destroy(session, ctf_trace); + } +} + +/* + * Cleanup a session. + */ +static void cleanup_session(struct relay_connection *conn, + struct relay_session *session) +{ + /* + * Very important that this is done before destroying the session so we + * can put back every viewer stream reference from the ctf_trace. + */ + destroy_viewer_streams_by_session(session); + try_destroy_streams(session); + cds_list_del(&session->viewer_session_list); + session_viewer_try_destroy(conn->sessions_ht, session); +} + /* * Send the viewer the list of current sessions. */ @@ -880,10 +948,6 @@ int viewer_get_new_streams(struct relay_connection *conn) send_streams = 1; response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_OK); - if (!send_streams) { - goto send_reply; - } - ret = make_viewer_streams(session, LTTNG_VIEWER_SEEK_LAST, NULL, &nb_unsent, &nb_created); if (ret < 0) { @@ -906,7 +970,7 @@ int viewer_get_new_streams(struct relay_connection *conn) * and try to destroy it. */ cds_list_del(&session->viewer_session_list); - session_viewer_try_destroy(conn->sessions_ht, session); + cleanup_session(conn, session); goto send_reply; } @@ -1014,10 +1078,6 @@ int viewer_attach_session(struct relay_connection *conn) goto send_reply; } - if (!send_streams) { - goto send_reply; - } - ret = make_viewer_streams(session, seek_type, &nb_streams, NULL, NULL); if (ret < 0) { goto end_unlock; @@ -1129,23 +1189,25 @@ int viewer_get_next_index(struct relay_connection *conn) rstream = stream_find_by_id(relay_streams_ht, vstream->stream_handle); assert(rstream); + pthread_mutex_lock(&rstream->viewer_stream_rotation_lock); if (!rstream->close_flag) { if (vstream->abort_flag) { /* Rotate on abort (overwrite). */ DBG("Viewer rotate because of overwrite"); ret = viewer_stream_rotate(vstream, rstream); if (ret < 0) { + pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock); goto end_unlock; } else if (ret == 1) { viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_HUP); viewer_stream_delete(vstream); viewer_stream_destroy(ctf_trace, vstream); + pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock); goto send_reply; } /* ret == 0 means successful so we continue. */ } - pthread_mutex_lock(&rstream->viewer_stream_rotation_lock); if (rstream->tracefile_count_current == vstream->tracefile_count_current) { if (rstream->beacon_ts_end != -1ULL && vstream->last_sent_index == rstream->total_index_received) { @@ -1166,24 +1228,25 @@ int viewer_get_next_index(struct relay_connection *conn) goto send_reply; } } - pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock); } else if (rstream->close_flag && vstream->close_write_flag && vstream->total_index_received == vstream->last_sent_index) { /* Last index sent and current tracefile closed in write */ viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_HUP); viewer_stream_delete(vstream); viewer_stream_destroy(ctf_trace, vstream); + pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock); goto send_reply; } else { vstream->close_write_flag = 1; } + pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock); if (!ctf_trace->metadata_received || ctf_trace->metadata_received > ctf_trace->metadata_sent) { viewer_index.flags |= LTTNG_VIEWER_FLAG_NEW_METADATA; } - ret = check_new_streams(vstream->session_id, conn->sessions_ht); + ret = check_new_streams(conn); if (ret < 0) { goto end_unlock; } else if (ret == 1) { @@ -1213,10 +1276,15 @@ int viewer_get_next_index(struct relay_connection *conn) sizeof(packet_index)); pthread_mutex_unlock(&vstream->overwrite_lock); if (ret < sizeof(packet_index)) { + unsigned int close_write_flag; + + pthread_mutex_lock(&rstream->viewer_stream_rotation_lock); + close_write_flag = vstream->close_write_flag; + pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock); /* * The tracefile is closed in write, so we read up to EOF. */ - if (vstream->close_write_flag == 1) { + if (close_write_flag == 1) { viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_RETRY); /* Rotate on normal EOF */ ret = viewer_stream_rotate(vstream, rstream); @@ -1352,7 +1420,7 @@ int viewer_get_packet(struct relay_connection *conn) goto send_reply; } - ret = check_new_streams(stream->session_id, conn->sessions_ht); + ret = check_new_streams(conn); if (ret < 0) { goto end_unlock; } else if (ret == 1) { @@ -1567,7 +1635,7 @@ int viewer_create_session(struct relay_connection *conn) DBG("Viewer create session received"); resp.status = htobe32(LTTNG_VIEWER_CREATE_SESSION_OK); - conn->viewer_session = zmalloc(sizeof(conn->viewer_session)); + conn->viewer_session = zmalloc(sizeof(*conn->viewer_session)); if (!conn->viewer_session) { ERR("Allocation viewer session"); resp.status = htobe32(LTTNG_VIEWER_CREATE_SESSION_ERR); @@ -1677,56 +1745,6 @@ void cleanup_connection_pollfd(struct lttng_poll_event *events, int pollfd) } } -/* - * Delete all streams for a specific session ID. - */ -static void destroy_viewer_streams_by_session(struct relay_session *session) -{ - struct relay_viewer_stream *stream; - struct lttng_ht_iter iter; - - assert(session); - - rcu_read_lock(); - cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, stream, - stream_n.node) { - struct ctf_trace *ctf_trace; - - health_code_update(); - if (stream->session_id != session->id) { - continue; - } - - ctf_trace = ctf_trace_find_by_path(session->ctf_traces_ht, - stream->path_name); - assert(ctf_trace); - - viewer_stream_delete(stream); - - if (stream->metadata_flag) { - ctf_trace->metadata_sent = 0; - ctf_trace->viewer_metadata_stream = NULL; - } - - viewer_stream_destroy(ctf_trace, stream); - } - rcu_read_unlock(); -} - -static void try_destroy_streams(struct relay_session *session) -{ - struct ctf_trace *ctf_trace; - struct lttng_ht_iter iter; - - assert(session); - - cds_lfht_for_each_entry(session->ctf_traces_ht->ht, &iter.iter, ctf_trace, - node.node) { - /* Attempt to destroy the ctf trace of that session. */ - ctf_trace_try_destroy(session, ctf_trace); - } -} - /* * Delete and destroy a connection. * @@ -1751,14 +1769,7 @@ static void destroy_connection(struct lttng_ht *relay_connections_ht, &conn->viewer_session->sessions_head, viewer_session_list) { DBG("Cleaning connection of session ID %" PRIu64, session->id); - /* - * Very important that this is done before destroying the session so we - * can put back every viewer stream reference from the ctf_trace. - */ - destroy_viewer_streams_by_session(session); - try_destroy_streams(session); - cds_list_del(&session->viewer_session_list); - session_viewer_try_destroy(conn->sessions_ht, session); + cleanup_session(conn, session); } rcu_read_unlock(); @@ -1844,7 +1855,7 @@ restart: health_code_update(); /* Thread quit pipe has been closed. Killing thread. */ - ret = check_live_conn_pipe(pollfd, revents); + ret = check_thread_quit_pipe(pollfd, revents); if (ret) { err = 0; goto exit;