}
/* Check and cleanup relayd */
+ rcu_read_lock();
relayd = consumer_find_relayd(stream->net_seq_idx);
if (relayd != NULL) {
- /* We are about to modify the relayd refcount */
- rcu_read_lock();
- if (!--relayd->refcount) {
+ uatomic_dec(&relayd->refcount);
+ assert(uatomic_read(&relayd->refcount) >= 0);
+ if (uatomic_read(&relayd->refcount) == 0) {
/* Refcount of the relayd struct is 0, destroy it */
consumer_destroy_relayd(relayd);
}
- rcu_read_unlock();
}
+ rcu_read_unlock();
if (!--stream->chan->refcount) {
free_chan = stream->chan;
pthread_mutex_lock(&consumer_data.lock);
/* Steal stream identifier, for UST */
consumer_steal_stream_key(stream->key);
- rcu_read_lock();
+ rcu_read_lock();
lttng_ht_lookup(consumer_data.stream_ht,
(void *)((unsigned long) stream->key), &iter);
node = lttng_ht_iter_get_node_ulong(&iter);
}
lttng_ht_add_unique_ulong(consumer_data.stream_ht, &stream->node);
- rcu_read_unlock();
/* Check and cleanup relayd */
relayd = consumer_find_relayd(stream->net_seq_idx);
if (relayd != NULL) {
- /* We are about to modify the relayd refcount */
- rcu_read_lock();
- relayd->refcount++;
- rcu_read_unlock();
+ uatomic_inc(&relayd->refcount);
}
+ rcu_read_unlock();
/* Update consumer data */
consumer_data.stream_count++;
* Find a relayd socket pair in the global consumer data.
*
* Return the object if found else NULL.
+ * RCU read-side lock must be held across this call and while using the
+ * returned object.
*/
struct consumer_relayd_sock_pair *consumer_find_relayd(int key)
{
goto error;
}
- rcu_read_lock();
-
lttng_ht_lookup(consumer_data.relayd_ht, (void *)((unsigned long) key),
&iter);
node = lttng_ht_iter_get_node_ulong(&iter);
relayd = caa_container_of(node, struct consumer_relayd_sock_pair, node);
}
- rcu_read_unlock();
-
error:
return relayd;
}
/* Reset data header */
memset(&data_hdr, 0, sizeof(data_hdr));
+ rcu_read_lock();
/* Get relayd reference of the stream. */
relayd = consumer_find_relayd(stream->net_seq_idx);
if (relayd == NULL) {
}
error:
+ rcu_read_unlock();
return outfd;
}
uint64_t metadata_id;
struct consumer_relayd_sock_pair *relayd = NULL;
+ /* RCU lock for the relayd pointer */
+ rcu_read_lock();
+
/* Flag that the current stream if set for network streaming. */
if (stream->net_seq_idx != -1) {
relayd = consumer_find_relayd(stream->net_seq_idx);
goto end;
}
- /* RCU lock for the relayd pointer */
- rcu_read_lock();
-
/* Handle stream on the relayd if the output is on the network */
if (relayd) {
/*
uint64_t metadata_id;
struct consumer_relayd_sock_pair *relayd = NULL;
+ /* RCU lock for the relayd pointer */
+ rcu_read_lock();
+
/* Flag that the current stream if set for network streaming. */
if (stream->net_seq_idx != -1) {
relayd = consumer_find_relayd(stream->net_seq_idx);
}
}
- /* RCU lock for the relayd pointer */
- rcu_read_lock();
-
/* Write metadata stream id before payload */
if (stream->metadata_flag && relayd) {
/*
return -ENOENT;
}
+ /* relayd needs RCU read-side protection */
+ rcu_read_lock();
+
switch (msg.cmd_type) {
case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
{
ret = write(ctx->consumer_poll_pipe[1], "", 1);
} while (ret < 0 && errno == EINTR);
end_nosignal:
+ rcu_read_unlock();
return 0;
}
uint64_t metadata_id;
struct consumer_relayd_sock_pair *relayd = NULL;
+ /* RCU lock for the relayd pointer */
+ rcu_read_lock();
+
/* Flag that the current stream if set for network streaming. */
if (stream->net_seq_idx != -1) {
relayd = consumer_find_relayd(stream->net_seq_idx);
if (relayd == NULL) {
+ ERR("Cannot find relay for network stream\n");
goto end;
}
}
if (relayd && stream->metadata_flag) {
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
}
+ rcu_read_unlock();
return written;
}
return -ENOENT;
}
+ /* relayd need RCU read-side lock */
+ rcu_read_lock();
+
switch (msg.cmd_type) {
case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
{
ret = write(ctx->consumer_poll_pipe[1], "", 1);
} while (ret < 0 && errno == EINTR);
end_nosignal:
+ rcu_read_unlock();
return 0;
}