projects
/
lttng-tools.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Fix: relayd: unbalanced RCU read lock/unlock
[lttng-tools.git]
/
src
/
bin
/
lttng-relayd
/
main.c
diff --git
a/src/bin/lttng-relayd/main.c
b/src/bin/lttng-relayd/main.c
index 60b6bf2214e8bee4d0266000a1db592982384940..24288d28756642fe3c74b68e04611326e9387007 100644
(file)
--- a/
src/bin/lttng-relayd/main.c
+++ b/
src/bin/lttng-relayd/main.c
@@
-44,6
+44,7
@@
#include <common/common.h>
#include <common/compat/poll.h>
#include <common/compat/socket.h>
#include <common/common.h>
#include <common/compat/poll.h>
#include <common/compat/socket.h>
+#include <common/compat/endian.h>
#include <common/defaults.h>
#include <common/daemonize.h>
#include <common/futex.h>
#include <common/defaults.h>
#include <common/daemonize.h>
#include <common/futex.h>
@@
-232,6
+233,11
@@
int set_option(int opt, const char *arg, const char *optname)
break;
case 'g':
tracing_group_name = strdup(arg);
break;
case 'g':
tracing_group_name = strdup(arg);
+ if (tracing_group_name == NULL) {
+ ret = -errno;
+ PERROR("strdup");
+ goto end;
+ }
tracing_group_name_override = 1;
break;
case 'h':
tracing_group_name_override = 1;
break;
case 'h':
@@
-250,7
+256,10
@@
int set_option(int opt, const char *arg, const char *optname)
if (arg) {
lttng_opt_verbose = config_parse_value(arg);
} else {
if (arg) {
lttng_opt_verbose = config_parse_value(arg);
} else {
- lttng_opt_verbose += 1;
+ /* Only 3 level of verbosity (-vvv). */
+ if (lttng_opt_verbose < 3) {
+ lttng_opt_verbose += 1;
+ }
}
break;
default:
}
break;
default:
@@
-382,8
+391,9
@@
int set_options(int argc, char **argv)
/* assign default values */
if (control_uri == NULL) {
/* assign default values */
if (control_uri == NULL) {
- ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
- DEFAULT_NETWORK_CONTROL_PORT);
+ ret = asprintf(&default_address,
+ "tcp://" DEFAULT_NETWORK_CONTROL_BIND_ADDRESS ":%d",
+ DEFAULT_NETWORK_CONTROL_PORT);
if (ret < 0) {
PERROR("asprintf default data address");
goto exit;
if (ret < 0) {
PERROR("asprintf default data address");
goto exit;
@@
-397,8
+407,9
@@
int set_options(int argc, char **argv)
}
}
if (data_uri == NULL) {
}
}
if (data_uri == NULL) {
- ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
- DEFAULT_NETWORK_DATA_PORT);
+ ret = asprintf(&default_address,
+ "tcp://" DEFAULT_NETWORK_DATA_BIND_ADDRESS ":%d",
+ DEFAULT_NETWORK_DATA_PORT);
if (ret < 0) {
PERROR("asprintf default data address");
goto exit;
if (ret < 0) {
PERROR("asprintf default data address");
goto exit;
@@
-412,8
+423,9
@@
int set_options(int argc, char **argv)
}
}
if (live_uri == NULL) {
}
}
if (live_uri == NULL) {
- ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
- DEFAULT_NETWORK_VIEWER_PORT);
+ ret = asprintf(&default_address,
+ "tcp://" DEFAULT_NETWORK_VIEWER_BIND_ADDRESS ":%d",
+ DEFAULT_NETWORK_VIEWER_PORT);
if (ret < 0) {
PERROR("asprintf default viewer control address");
goto exit;
if (ret < 0) {
PERROR("asprintf default viewer control address");
goto exit;
@@
-1198,6
+1210,7
@@
int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr,
stream->session_id = session->id;
stream->index_fd = -1;
stream->read_index_fd = -1;
stream->session_id = session->id;
stream->index_fd = -1;
stream->read_index_fd = -1;
+ stream->ctf_stream_id = -1ULL;
lttng_ht_node_init_u64(&stream->node, stream->stream_handle);
pthread_mutex_init(&stream->lock, NULL);
lttng_ht_node_init_u64(&stream->node, stream->stream_handle);
pthread_mutex_init(&stream->lock, NULL);
@@
-1934,9
+1947,10
@@
int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
DBG("Received live beacon for stream %" PRIu64, stream->stream_handle);
/*
DBG("Received live beacon for stream %" PRIu64, stream->stream_handle);
/*
- * Only flag a stream inactive when it has already received data.
+ * Only flag a stream inactive when it has already received data
+ * and no indexes are in flight.
*/
*/
- if (stream->total_index_received > 0) {
+ if (stream->total_index_received > 0
&& stream->indexes_in_flight == 0
) {
stream->beacon_ts_end = be64toh(index_info.timestamp_end);
}
ret = 0;
stream->beacon_ts_end = be64toh(index_info.timestamp_end);
}
ret = 0;
@@
-1953,9
+1967,13
@@
int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
goto end_rcu_unlock;
}
index_created = 1;
goto end_rcu_unlock;
}
index_created = 1;
+ stream->indexes_in_flight++;
}
copy_index_control_data(index, &index_info);
}
copy_index_control_data(index, &index_info);
+ if (stream->ctf_stream_id == -1ULL) {
+ stream->ctf_stream_id = be64toh(index_info.stream_id);
+ }
if (index_created) {
/*
if (index_created) {
/*
@@
-1980,6
+1998,8
@@
int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
goto end_rcu_unlock;
}
stream->total_index_received++;
goto end_rcu_unlock;
}
stream->total_index_received++;
+ stream->indexes_in_flight--;
+ assert(stream->indexes_in_flight >= 0);
}
end_rcu_unlock:
}
end_rcu_unlock:
@@
-2143,6
+2163,7
@@
static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num,
goto error;
}
index_created = 1;
goto error;
}
index_created = 1;
+ stream->indexes_in_flight++;
}
if (rotate_index || stream->index_fd < 0) {
}
if (rotate_index || stream->index_fd < 0) {
@@
-2185,6
+2206,8
@@
static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num,
goto error;
}
stream->total_index_received++;
goto error;
}
stream->total_index_received++;
+ stream->indexes_in_flight--;
+ assert(stream->indexes_in_flight >= 0);
}
error:
}
error:
@@
-2386,7
+2409,7
@@
static void destroy_connection(struct lttng_ht *relay_connections_ht,
connection_delete(relay_connections_ht, conn);
/* For the control socket, we try to destroy the session. */
connection_delete(relay_connections_ht, conn);
/* For the control socket, we try to destroy the session. */
- if (conn->type == RELAY_CONTROL) {
+ if (conn->type == RELAY_CONTROL
&& conn->session
) {
destroy_session(conn->session, conn->sessions_ht);
}
destroy_session(conn->session, conn->sessions_ht);
}
@@
-2408,6
+2431,7
@@
void *relay_thread_worker(void *data)
struct lttcomm_relayd_hdr recv_hdr;
struct relay_local_data *relay_ctx = (struct relay_local_data *) data;
struct lttng_ht *sessions_ht = relay_ctx->sessions_ht;
struct lttcomm_relayd_hdr recv_hdr;
struct relay_local_data *relay_ctx = (struct relay_local_data *) data;
struct lttng_ht *sessions_ht = relay_ctx->sessions_ht;
+ struct relay_index *index;
DBG("[thread] Relay worker started");
DBG("[thread] Relay worker started");
@@
-2596,6
+2620,7
@@
restart:
if (revents & LPOLLIN) {
if (conn->type != RELAY_DATA) {
if (revents & LPOLLIN) {
if (conn->type != RELAY_DATA) {
+ rcu_read_unlock();
continue;
}
continue;
}
@@
-2639,6
+2664,14
@@
error:
}
rcu_read_unlock();
error_poll_create:
}
rcu_read_unlock();
error_poll_create:
+ rcu_read_lock();
+ cds_lfht_for_each_entry(indexes_ht->ht, &iter.iter, index,
+ index_n.node) {
+ health_code_update();
+ relay_index_delete(index);
+ relay_index_free_safe(index);
+ }
+ rcu_read_unlock();
lttng_ht_destroy(indexes_ht);
indexes_ht_error:
lttng_ht_destroy(relay_connections_ht);
lttng_ht_destroy(indexes_ht);
indexes_ht_error:
lttng_ht_destroy(relay_connections_ht);
This page took
0.026573 seconds
and
4
git commands to generate.