X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Flive.c;h=06c1c15134cf63af055729b6b28f75d0489962ce;hp=831d760ae57baa2d178fb4e2378ee6387d234968;hb=b5a6470f372799b28d3d20603c1c0c8e5871dd63;hpb=d227d5bd12acbff5cb7e703f9bed139e066a2033 diff --git a/src/bin/lttng-relayd/live.c b/src/bin/lttng-relayd/live.c index 831d760ae..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; @@ -946,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) { @@ -1080,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; @@ -1195,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) { @@ -1232,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) { @@ -1279,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); @@ -1418,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) { @@ -1633,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); @@ -1853,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;