projects
/
lttng-tools.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Fix: initialize context list node on creation
[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 ac03cf846a7ec7992d7c21f9423d0af6813f22db..802d7e107abf019720f95b52fabdf556d68df9fb 100644
(file)
--- a/
src/bin/lttng-relayd/main.c
+++ b/
src/bin/lttng-relayd/main.c
@@
-882,11
+882,12
@@
restart:
new_conn->sock = newsock;
/* Enqueue request for the dispatcher thread. */
new_conn->sock = newsock;
/* Enqueue request for the dispatcher thread. */
- cds_wfq_enqueue(&relay_conn_queue.queue, &new_conn->qnode);
+ cds_wfcq_enqueue(&relay_conn_queue.head, &relay_conn_queue.tail,
+ &new_conn->qnode);
/*
* Wake the dispatch queue futex. Implicit memory barrier with
/*
* Wake the dispatch queue futex. Implicit memory barrier with
- * the exchange in cds_wfq_enqueue.
+ * the exchange in cds_wf
c
q_enqueue.
*/
futex_nto1_wake(&relay_conn_queue.futex);
}
*/
futex_nto1_wake(&relay_conn_queue.futex);
}
@@
-933,7
+934,7
@@
void *relay_thread_dispatcher(void *data)
{
int err = -1;
ssize_t ret;
{
int err = -1;
ssize_t ret;
- struct cds_wfq_node *node;
+ struct cds_wf
c
q_node *node;
struct relay_connection *new_conn = NULL;
DBG("[thread] Relay dispatcher started");
struct relay_connection *new_conn = NULL;
DBG("[thread] Relay dispatcher started");
@@
-956,7
+957,8
@@
void *relay_thread_dispatcher(void *data)
health_code_update();
/* Dequeue commands */
health_code_update();
/* Dequeue commands */
- node = cds_wfq_dequeue_blocking(&relay_conn_queue.queue);
+ node = cds_wfcq_dequeue_blocking(&relay_conn_queue.head,
+ &relay_conn_queue.tail);
if (node == NULL) {
DBG("Woken up but nothing in the relay command queue");
/* Continue thread execution */
if (node == NULL) {
DBG("Woken up but nothing in the relay command queue");
/* Continue thread execution */
@@
-1205,6
+1207,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);
@@
-1342,7
+1345,6
@@
int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr,
stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num);
stream->close_flag = 1;
session->stream_count--;
stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num);
stream->close_flag = 1;
session->stream_count--;
- assert(session->stream_count >= 0);
/* Check if we can close it or else the data will do it. */
try_close_stream(session, stream);
/* Check if we can close it or else the data will do it. */
try_close_stream(session, stream);
@@
-1941,9
+1943,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;
@@
-1960,9
+1963,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) {
/*
@@
-1987,6
+1994,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:
@@
-2150,6
+2159,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) {
@@
-2192,6
+2202,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:
@@
-2758,7
+2770,7
@@
int main(int argc, char **argv)
}
/* Init relay command queue. */
}
/* Init relay command queue. */
- cds_wf
q_init(&relay_conn_queue.queue
);
+ cds_wf
cq_init(&relay_conn_queue.head, &relay_conn_queue.tail
);
/* Set up max poll set size */
lttng_poll_set_max_size();
/* Set up max poll set size */
lttng_poll_set_max_size();
This page took
0.059253 seconds
and
4
git commands to generate.