The consumer's streams refer to a 'net_seq_idx' of which the
meaning must have been lost in the sands of time. It is a
unique identifier of a given relay daemon. Hence, renaming it to
'relayd_id' appears sensible.
Signed-off-by: Jérémie Galarneau <jeremie.galarneau@efficios.com>
Signed-off-by: Jonathan Rajotte <jonathan.rajotte-julien@efficios.com>
stream->relayd_stream_id,
stream->next_net_seq_num - 1);
if (ret < 0) {
stream->relayd_stream_id,
stream->next_net_seq_num - 1);
if (ret < 0) {
- ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
+ ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64 ".", relayd->id);
lttng_consumer_cleanup_relayd(relayd);
}
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
lttng_consumer_cleanup_relayd(relayd);
}
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
uatomic_read(&relayd->destroy_flag)) {
consumer_destroy_relayd(relayd);
}
uatomic_read(&relayd->destroy_flag)) {
consumer_destroy_relayd(relayd);
}
- stream->net_seq_idx = (uint64_t) -1ULL;
+ stream->relayd_id = (uint64_t) -1ULL;
stream->sent_to_relayd = 0;
}
stream->sent_to_relayd = 0;
}
/* Check and cleanup relayd if needed. */
rcu_read_lock();
/* Check and cleanup relayd if needed. */
rcu_read_lock();
- relayd = consumer_find_relayd(stream->net_seq_idx);
+ relayd = consumer_find_relayd(stream->relayd_id);
if (relayd != NULL) {
consumer_stream_relayd_close(stream, relayd);
}
if (relayd != NULL) {
consumer_stream_relayd_close(stream, relayd);
}
assert(element);
rcu_read_lock();
assert(element);
rcu_read_lock();
- if (stream->net_seq_idx != (uint64_t) -1ULL) {
+ if (stream->relayd_id != (uint64_t) -1ULL) {
struct consumer_relayd_sock_pair *relayd;
struct consumer_relayd_sock_pair *relayd;
- relayd = consumer_find_relayd(stream->net_seq_idx);
+ relayd = consumer_find_relayd(stream->relayd_id);
if (relayd) {
pthread_mutex_lock(&relayd->ctrl_sock_mutex);
ret = relayd_send_index(&relayd->control_sock, element,
if (relayd) {
pthread_mutex_lock(&relayd->ctrl_sock_mutex);
ret = relayd_send_index(&relayd->control_sock, element,
* Communication error with lttng-relayd,
* perform cleanup now
*/
* Communication error with lttng-relayd,
* perform cleanup now
*/
- ERR("Relayd send index failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
+ ERR("Relayd send index failed. Cleaning up relayd %" PRIu64 ".", relayd->id);
lttng_consumer_cleanup_relayd(relayd);
ret = -1;
}
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
} else {
ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't write index.",
lttng_consumer_cleanup_relayd(relayd);
ret = -1;
}
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
} else {
ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't write index.",
- stream->key, stream->net_seq_idx);
+ stream->key, stream->relayd_id);
- * Update the end point status of all streams having the given network sequence
- * index (relayd index).
+ * Update the end point status of all streams having the given relayd id.
*
* It's atomically set without having the stream mutex locked which is fine
* because we handle the write/read race with a pipe wakeup for each thread.
*/
*
* It's atomically set without having the stream mutex locked which is fine
* because we handle the write/read race with a pipe wakeup for each thread.
*/
-static void update_endpoint_status_by_netidx(uint64_t net_seq_idx,
+static void update_endpoint_status_by_netidx(uint64_t relayd_id,
enum consumer_endpoint_status status)
{
struct lttng_ht_iter iter;
struct lttng_consumer_stream *stream;
enum consumer_endpoint_status status)
{
struct lttng_ht_iter iter;
struct lttng_consumer_stream *stream;
- DBG("Consumer set delete flag on stream by idx %" PRIu64, net_seq_idx);
+ DBG("Consumer set delete flag on stream by idx %" PRIu64, relayd_id);
rcu_read_lock();
/* Let's begin with metadata */
cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
rcu_read_lock();
/* Let's begin with metadata */
cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
- if (stream->net_seq_idx == net_seq_idx) {
+ if (stream->relayd_id == relayd_id) {
uatomic_set(&stream->endpoint_status, status);
DBG("Delete flag set to metadata stream %d", stream->wait_fd);
}
uatomic_set(&stream->endpoint_status, status);
DBG("Delete flag set to metadata stream %d", stream->wait_fd);
}
/* Follow up by the data streams */
cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
/* Follow up by the data streams */
cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
- if (stream->net_seq_idx == net_seq_idx) {
+ if (stream->relayd_id == relayd_id) {
uatomic_set(&stream->endpoint_status, status);
DBG("Delete flag set to data stream %d", stream->wait_fd);
}
uatomic_set(&stream->endpoint_status, status);
DBG("Delete flag set to data stream %d", stream->wait_fd);
}
- DBG("Cleaning up relayd object ID %"PRIu64, relayd->net_seq_idx);
+ DBG("Cleaning up relayd object ID %"PRIu64, relayd->id);
/* Save the net sequence index before destroying the object */
/* Save the net sequence index before destroying the object */
- netidx = relayd->net_seq_idx;
/*
* Delete the relayd from the relayd hash table, close the sockets and free
/*
* Delete the relayd from the relayd hash table, close the sockets and free
stream->state = state;
stream->uid = uid;
stream->gid = gid;
stream->state = state;
stream->uid = uid;
stream->gid = gid;
- stream->net_seq_idx = relayd_id;
+ stream->relayd_id = relayd_id;
stream->session_id = session_id;
stream->monitor = monitor;
stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
stream->session_id = session_id;
stream->monitor = monitor;
stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
" relayd_id %" PRIu64 ", session_id %" PRIu64,
stream->name, stream->key, channel_key,
DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
" relayd_id %" PRIu64 ", session_id %" PRIu64,
stream->name, stream->key, channel_key,
- stream->net_seq_idx, stream->session_id);
+ stream->relayd_id, stream->session_id);
rcu_read_unlock();
return stream;
rcu_read_unlock();
return stream;
assert(relayd);
lttng_ht_lookup(consumer_data.relayd_ht,
assert(relayd);
lttng_ht_lookup(consumer_data.relayd_ht,
- &relayd->net_seq_idx, &iter);
node = lttng_ht_iter_get_node_u64(&iter);
if (node != NULL) {
goto end;
node = lttng_ht_iter_get_node_u64(&iter);
if (node != NULL) {
goto end;
* Allocate and return a consumer relayd socket.
*/
static struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
* Allocate and return a consumer relayd socket.
*/
static struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
{
struct consumer_relayd_sock_pair *obj = NULL;
/* net sequence index of -1 is a failure */
{
struct consumer_relayd_sock_pair *obj = NULL;
/* net sequence index of -1 is a failure */
- if (net_seq_idx == (uint64_t) -1ULL) {
+ if (relayd_id == (uint64_t) -1ULL) {
- obj->net_seq_idx = net_seq_idx;
obj->refcount = 0;
obj->destroy_flag = 0;
obj->control_sock.sock.fd = -1;
obj->data_sock.sock.fd = -1;
obj->refcount = 0;
obj->destroy_flag = 0;
obj->control_sock.sock.fd = -1;
obj->data_sock.sock.fd = -1;
- lttng_ht_node_init_u64(&obj->node, obj->net_seq_idx);
+ lttng_ht_node_init_u64(&obj->node, obj->id);
pthread_mutex_init(&obj->ctrl_sock_mutex, NULL);
error:
pthread_mutex_init(&obj->ctrl_sock_mutex, NULL);
error:
struct consumer_relayd_sock_pair *relayd;
assert(stream);
struct consumer_relayd_sock_pair *relayd;
assert(stream);
- assert(stream->net_seq_idx != -1ULL);
+ assert(stream->relayd_id != -1ULL);
assert(path);
/* The stream is not metadata. Get relayd reference if exists. */
rcu_read_lock();
assert(path);
/* The stream is not metadata. Get relayd reference if exists. */
rcu_read_lock();
- relayd = consumer_find_relayd(stream->net_seq_idx);
+ relayd = consumer_find_relayd(stream->relayd_id);
if (relayd != NULL) {
/* Add stream on the relayd */
pthread_mutex_lock(&relayd->ctrl_sock_mutex);
if (relayd != NULL) {
/* Add stream on the relayd */
pthread_mutex_lock(&relayd->ctrl_sock_mutex);
stream->chan->tracefile_size, stream->chan->tracefile_count);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
stream->chan->tracefile_size, stream->chan->tracefile_count);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
- ERR("Relayd add stream failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+ ERR("Relayd add stream failed. Cleaning up relayd %" PRIu64".", relayd->id);
lttng_consumer_cleanup_relayd(relayd);
goto end;
}
lttng_consumer_cleanup_relayd(relayd);
goto end;
}
stream->sent_to_relayd = 1;
} else {
ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't send it.",
stream->sent_to_relayd = 1;
} else {
ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't send it.",
- stream->key, stream->net_seq_idx);
+ stream->key, stream->relayd_id);
ret = -1;
goto end;
}
DBG("Stream %s with key %" PRIu64 " sent to relayd id %" PRIu64,
ret = -1;
goto end;
}
DBG("Stream %s with key %" PRIu64 " sent to relayd id %" PRIu64,
- stream->name, stream->key, stream->net_seq_idx);
+ stream->name, stream->key, stream->relayd_id);
*
* Returns 0 on success, < 0 on error
*/
*
* Returns 0 on success, < 0 on error
*/
-int consumer_send_relayd_streams_sent(uint64_t net_seq_idx)
+int consumer_send_relayd_streams_sent(uint64_t relayd_id)
{
int ret = 0;
struct consumer_relayd_sock_pair *relayd;
{
int ret = 0;
struct consumer_relayd_sock_pair *relayd;
- assert(net_seq_idx != -1ULL);
+ assert(relayd_id != -1ULL);
/* The stream is not metadata. Get relayd reference if exists. */
rcu_read_lock();
/* The stream is not metadata. Get relayd reference if exists. */
rcu_read_lock();
- relayd = consumer_find_relayd(net_seq_idx);
+ relayd = consumer_find_relayd(relayd_id);
if (relayd != NULL) {
/* Add stream on the relayd */
pthread_mutex_lock(&relayd->ctrl_sock_mutex);
ret = relayd_streams_sent(&relayd->control_sock);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
if (relayd != NULL) {
/* Add stream on the relayd */
pthread_mutex_lock(&relayd->ctrl_sock_mutex);
ret = relayd_streams_sent(&relayd->control_sock);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
- ERR("Relayd streams sent failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+ ERR("Relayd streams sent failed. Cleaning up relayd %" PRIu64".", relayd->id);
lttng_consumer_cleanup_relayd(relayd);
goto end;
}
} else {
ERR("Relayd ID %" PRIu64 " unknown. Can't send streams_sent.",
lttng_consumer_cleanup_relayd(relayd);
goto end;
}
} else {
ERR("Relayd ID %" PRIu64 " unknown. Can't send streams_sent.",
ret = -1;
goto end;
}
ret = 0;
ret = -1;
goto end;
}
ret = 0;
- DBG("All streams sent relayd id %" PRIu64, net_seq_idx);
+ DBG("All streams sent relayd id %" PRIu64, relayd_id);
/* The stream is not metadata. Get relayd reference if exists. */
rcu_read_lock();
/* The stream is not metadata. Get relayd reference if exists. */
rcu_read_lock();
- relayd = consumer_find_relayd(stream->net_seq_idx);
+ relayd = consumer_find_relayd(stream->relayd_id);
if (relayd) {
consumer_stream_relayd_close(stream, relayd);
}
if (relayd) {
consumer_stream_relayd_close(stream, relayd);
}
rcu_read_lock();
/* Flag that the current stream if set for network streaming. */
rcu_read_lock();
/* Flag that the current stream if set for network streaming. */
- if (stream->net_seq_idx != (uint64_t) -1ULL) {
- relayd = consumer_find_relayd(stream->net_seq_idx);
+ if (stream->relayd_id != (uint64_t) -1ULL) {
+ relayd = consumer_find_relayd(stream->relayd_id);
if (relayd == NULL) {
ret = -EPIPE;
goto end;
if (relayd == NULL) {
ret = -EPIPE;
goto end;
* cleanup the relayd object and all associated streams.
*/
if (relayd && relayd_hang_up) {
* cleanup the relayd object and all associated streams.
*/
if (relayd && relayd_hang_up) {
- ERR("Relayd hangup. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+ ERR("Relayd hangup. Cleaning up relayd %" PRIu64".", relayd->id);
lttng_consumer_cleanup_relayd(relayd);
}
lttng_consumer_cleanup_relayd(relayd);
}
rcu_read_lock();
/* Flag that the current stream if set for network streaming. */
rcu_read_lock();
/* Flag that the current stream if set for network streaming. */
- if (stream->net_seq_idx != (uint64_t) -1ULL) {
- relayd = consumer_find_relayd(stream->net_seq_idx);
+ if (stream->relayd_id != (uint64_t) -1ULL) {
+ relayd = consumer_find_relayd(stream->relayd_id);
if (relayd == NULL) {
written = -ret;
goto end;
if (relayd == NULL) {
written = -ret;
goto end;
* cleanup the relayd object and all associated streams.
*/
if (relayd && relayd_hang_up) {
* cleanup the relayd object and all associated streams.
*/
if (relayd && relayd_hang_up) {
- ERR("Relayd hangup. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+ ERR("Relayd hangup. Cleaning up relayd %" PRIu64".", relayd->id);
lttng_consumer_cleanup_relayd(relayd);
/* Skip splice error so the consumer does not fail */
goto end;
lttng_consumer_cleanup_relayd(relayd);
/* Skip splice error so the consumer does not fail */
goto end;
* This will create a relayd socket pair and add it to the relayd hash table.
* The caller MUST acquire a RCU read side lock before calling it.
*/
* This will create a relayd socket pair and add it to the relayd hash table.
* The caller MUST acquire a RCU read side lock before calling it.
*/
- void consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
+ void consumer_add_relayd_socket(uint64_t relayd_id, int sock_type,
struct lttng_consumer_local_data *ctx, int sock,
struct pollfd *consumer_sockpoll,
struct lttcomm_relayd_sock *relayd_sock, uint64_t sessiond_id,
struct lttng_consumer_local_data *ctx, int sock,
struct pollfd *consumer_sockpoll,
struct lttcomm_relayd_sock *relayd_sock, uint64_t sessiond_id,
assert(ctx);
assert(relayd_sock);
assert(ctx);
assert(relayd_sock);
- DBG("Consumer adding relayd socket (idx: %" PRIu64 ")", net_seq_idx);
+ DBG("Consumer adding relayd socket (idx: %" PRIu64 ")", relayd_id);
/* Get relayd reference if exists. */
/* Get relayd reference if exists. */
- relayd = consumer_find_relayd(net_seq_idx);
+ relayd = consumer_find_relayd(relayd_id);
if (relayd == NULL) {
assert(sock_type == LTTNG_STREAM_CONTROL);
/* Not found. Allocate one. */
if (relayd == NULL) {
assert(sock_type == LTTNG_STREAM_CONTROL);
/* Not found. Allocate one. */
- relayd = consumer_allocate_relayd_sock_pair(net_seq_idx);
+ relayd = consumer_allocate_relayd_sock_pair(relayd_id);
if (relayd == NULL) {
ret = -ENOMEM;
ret_code = LTTCOMM_CONSUMERD_ENOMEM;
if (relayd == NULL) {
ret = -ENOMEM;
ret_code = LTTCOMM_CONSUMERD_ENOMEM;
DBG("Consumer %s socket created successfully with net idx %" PRIu64 " (fd: %d)",
sock_type == LTTNG_STREAM_CONTROL ? "control" : "data",
DBG("Consumer %s socket created successfully with net idx %" PRIu64 " (fd: %d)",
sock_type == LTTNG_STREAM_CONTROL ? "control" : "data",
- relayd->net_seq_idx, fd);
/* We successfully added the socket. Send status back. */
ret = consumer_send_status_msg(sock, ret_code);
/* We successfully added the socket. Send status back. */
ret = consumer_send_status_msg(sock, ret_code);
struct lttng_ht_iter iter;
struct consumer_relayd_sock_pair *relayd = NULL;
struct lttng_ht_iter iter;
struct consumer_relayd_sock_pair *relayd = NULL;
- /* Iterate over all relayd since they are indexed by net_seq_idx. */
+ /* Iterate over all relayd since they are indexed by relayd_id. */
cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd,
node.node) {
/*
cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd,
node.node) {
/*
if (ret < 0) {
/* Communication error thus the relayd so no data pending. */
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
/* Communication error thus the relayd so no data pending. */
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
- ERR("Relayd begin data pending failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+ ERR("Relayd begin data pending failed. Cleaning up relayd %" PRIu64".", relayd->id);
lttng_consumer_cleanup_relayd(relayd);
goto data_not_pending;
}
lttng_consumer_cleanup_relayd(relayd);
goto data_not_pending;
}
goto data_pending;
}
if (ret < 0) {
goto data_pending;
}
if (ret < 0) {
- ERR("Relayd data pending failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+ ERR("Relayd data pending failed. Cleaning up relayd %" PRIu64".", relayd->id);
lttng_consumer_cleanup_relayd(relayd);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
pthread_mutex_unlock(&stream->lock);
lttng_consumer_cleanup_relayd(relayd);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
pthread_mutex_unlock(&stream->lock);
relayd->relayd_session_id, &is_data_inflight);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
relayd->relayd_session_id, &is_data_inflight);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
- ERR("Relayd end data pending failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+ ERR("Relayd end data pending failed. Cleaning up relayd %" PRIu64".", relayd->id);
lttng_consumer_cleanup_relayd(relayd);
goto data_not_pending;
}
lttng_consumer_cleanup_relayd(relayd);
goto data_not_pending;
}
/* UID/GID of the user owning the session to which stream belongs */
uid_t uid;
gid_t gid;
/* UID/GID of the user owning the session to which stream belongs */
uid_t uid;
gid_t gid;
- /* Network sequence number. Indicating on which relayd socket it goes. */
- uint64_t net_seq_idx;
+ /*
+ * Relayd id, indicating on which relayd socket it goes. Set to -1ULL if
+ * not the stream is not associated to a relay daemon.
+ */
+ uint64_t relayd_id;
/*
* Indicate if this stream was successfully sent to a relayd. This is set
* after the refcount of the relayd is incremented and is checked when the
/*
* Indicate if this stream was successfully sent to a relayd. This is set
* after the refcount of the relayd is incremented and is checked when the
*/
struct consumer_relayd_sock_pair {
/* Network sequence number. */
*/
struct consumer_relayd_sock_pair {
/* Network sequence number. */
/* Number of stream associated with this relayd */
int refcount;
/* Number of stream associated with this relayd */
int refcount;
/* lttng-relayd consumer command */
struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key);
int consumer_send_relayd_stream(struct lttng_consumer_stream *stream, char *path);
/* lttng-relayd consumer command */
struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key);
int consumer_send_relayd_stream(struct lttng_consumer_stream *stream, char *path);
-int consumer_send_relayd_streams_sent(uint64_t net_seq_idx);
+int consumer_send_relayd_streams_sent(uint64_t relayd_id);
void close_relayd_stream(struct lttng_consumer_stream *stream);
struct lttng_consumer_channel *consumer_find_channel(uint64_t key);
int consumer_handle_stream_before_relayd(struct lttng_consumer_stream *stream,
void close_relayd_stream(struct lttng_consumer_stream *stream);
struct lttng_consumer_channel *consumer_find_channel(uint64_t key);
int consumer_handle_stream_before_relayd(struct lttng_consumer_stream *stream,
ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
struct lttng_consumer_local_data *ctx);
int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream);
ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
struct lttng_consumer_local_data *ctx);
int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream);
-void consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
+void consumer_add_relayd_socket(uint64_t relayd_id, int sock_type,
struct lttng_consumer_local_data *ctx, int sock,
struct pollfd *consumer_sockpoll, struct lttcomm_relayd_sock *relayd_sock,
uint64_t sessiond_id, uint64_t relayd_session_id);
struct lttng_consumer_local_data *ctx, int sock,
struct pollfd *consumer_sockpoll, struct lttcomm_relayd_sock *relayd_sock,
uint64_t sessiond_id, uint64_t relayd_session_id);
* Assign the received relayd ID so we can use it for streaming. The streams
* are not visible to anyone so this is OK to change it.
*/
* Assign the received relayd ID so we can use it for streaming. The streams
* are not visible to anyone so this is OK to change it.
*/
- stream->net_seq_idx = relayd_id;
+ stream->relayd_id = relayd_id;
channel->relayd_id = relayd_id;
if (relayd_id != (uint64_t) -1ULL) {
ret = consumer_send_relayd_stream(stream, path);
channel->relayd_id = relayd_id;
if (relayd_id != (uint64_t) -1ULL) {
ret = consumer_send_relayd_stream(stream, path);
}
} else {
close_relayd_stream(stream);
}
} else {
close_relayd_stream(stream);
- stream->net_seq_idx = (uint64_t) -1ULL;
+ stream->relayd_id = (uint64_t) -1ULL;
}
pthread_mutex_unlock(&stream->lock);
}
}
pthread_mutex_unlock(&stream->lock);
}
if (use_relayd) {
close_relayd_stream(metadata_stream);
if (use_relayd) {
close_relayd_stream(metadata_stream);
- metadata_stream->net_seq_idx = (uint64_t) -1ULL;
+ metadata_stream->relayd_id = (uint64_t) -1ULL;
} else {
if (metadata_stream->out_fd >= 0) {
ret = close(metadata_stream->out_fd);
} else {
if (metadata_stream->out_fd >= 0) {
ret = close(metadata_stream->out_fd);
if (!channel->monitor) {
DBG("Kernel consumer add stream %s in no monitor mode with "
"relayd id %" PRIu64, new_stream->name,
if (!channel->monitor) {
DBG("Kernel consumer add stream %s in no monitor mode with "
"relayd id %" PRIu64, new_stream->name,
- new_stream->net_seq_idx);
+ new_stream->relayd_id);
cds_list_add(&new_stream->send_node, &channel->streams.head);
break;
}
/* Send stream to relayd if the stream has an ID. */
cds_list_add(&new_stream->send_node, &channel->streams.head);
break;
}
/* Send stream to relayd if the stream has an ID. */
- if (new_stream->net_seq_idx != (uint64_t) -1ULL) {
+ if (new_stream->relayd_id != (uint64_t) -1ULL) {
ret = consumer_send_relayd_stream(new_stream,
new_stream->chan->pathname);
if (ret < 0) {
ret = consumer_send_relayd_stream(new_stream,
new_stream->chan->pathname);
if (ret < 0) {
*/
if (channel->streams_sent_to_relayd) {
ret = consumer_send_relayd_streams_sent(
*/
if (channel->streams_sent_to_relayd) {
ret = consumer_send_relayd_streams_sent(
- new_stream->net_seq_idx);
+ new_stream->relayd_id);
if (ret < 0) {
goto end_nosignal;
}
if (ret < 0) {
goto end_nosignal;
}
* network streaming or the full padding (len) size when we are _not_
* streaming.
*/
* network streaming or the full padding (len) size when we are _not_
* streaming.
*/
- if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) ||
- (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) {
+ if ((ret != subbuf_size && stream->relayd_id != (uint64_t) -1ULL) ||
+ (ret != len && stream->relayd_id == (uint64_t) -1ULL)) {
/*
* Display the error but continue processing to try to release the
* subbuffer. This is a DBG statement since this is possible to
/*
* Display the error but continue processing to try to release the
* subbuffer. This is a DBG statement since this is possible to
* Don't create anything if this is set for streaming or should not be
* monitored.
*/
* Don't create anything if this is set for streaming or should not be
* monitored.
*/
- if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) {
+ if (stream->relayd_id == (uint64_t) -1ULL && stream->chan->monitor) {
ret = utils_create_stream_file(stream->chan->pathname, stream->name,
stream->chan->tracefile_size, stream->tracefile_count_current,
stream->uid, stream->gid, NULL);
ret = utils_create_stream_file(stream->chan->pathname, stream->name,
stream->chan->tracefile_size, stream->tracefile_count_current,
stream->uid, stream->gid, NULL);
{
int ret, ret_code = LTTCOMM_CONSUMERD_SUCCESS;
struct lttng_consumer_stream *stream;
{
int ret, ret_code = LTTCOMM_CONSUMERD_SUCCESS;
struct lttng_consumer_stream *stream;
- uint64_t net_seq_idx = -1ULL;
+ uint64_t relayd_id = -1ULL;
assert(channel);
assert(ctx);
assert(channel);
assert(ctx);
}
ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
}
}
ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
}
- if (net_seq_idx == -1ULL) {
- net_seq_idx = stream->net_seq_idx;
+ if (relayd_id == -1ULL) {
+ relayd_id = stream->relayd_id;
}
/* Send metadata stream to relayd if needed. */
}
/* Send metadata stream to relayd if needed. */
- if (metadata->metadata_stream->net_seq_idx != (uint64_t) -1ULL) {
+ if (metadata->metadata_stream->relayd_id != (uint64_t) -1ULL) {
ret = consumer_send_relayd_stream(metadata->metadata_stream,
metadata->pathname);
if (ret < 0) {
ret = consumer_send_relayd_stream(metadata->metadata_stream,
metadata->pathname);
if (ret < 0) {
goto error;
}
ret = consumer_send_relayd_streams_sent(
goto error;
}
ret = consumer_send_relayd_streams_sent(
- metadata->metadata_stream->net_seq_idx);
+ metadata->metadata_stream->relayd_id);
if (ret < 0) {
ret = LTTCOMM_CONSUMERD_RELAYD_FAIL;
goto error;
if (ret < 0) {
ret = LTTCOMM_CONSUMERD_RELAYD_FAIL;
goto error;
assert(metadata_stream);
if (relayd_id != (uint64_t) -1ULL) {
assert(metadata_stream);
if (relayd_id != (uint64_t) -1ULL) {
- metadata_stream->net_seq_idx = relayd_id;
+ metadata_stream->relayd_id = relayd_id;
ret = consumer_send_relayd_stream(metadata_stream, path);
if (ret < 0) {
goto error_stream;
ret = consumer_send_relayd_stream(metadata_stream, path);
if (ret < 0) {
goto error_stream;
/* Lock stream because we are about to change its state. */
pthread_mutex_lock(&stream->lock);
/* Lock stream because we are about to change its state. */
pthread_mutex_lock(&stream->lock);
- stream->net_seq_idx = relayd_id;
+ stream->relayd_id = relayd_id;
if (use_relayd) {
ret = consumer_send_relayd_stream(stream, path);
if (use_relayd) {
ret = consumer_send_relayd_stream(stream, path);
* The mmap operation should write subbuf_size amount of data when network
* streaming or the full padding (len) size when we are _not_ streaming.
*/
* The mmap operation should write subbuf_size amount of data when network
* streaming or the full padding (len) size when we are _not_ streaming.
*/
- if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) ||
- (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) {
+ if ((ret != subbuf_size && stream->relayd_id != (uint64_t) -1ULL) ||
+ (ret != len && stream->relayd_id == (uint64_t) -1ULL)) {
/*
* Display the error but continue processing to try to release the
* subbuffer. This is a DBG statement since any unexpected kill or
/*
* Display the error but continue processing to try to release the
* subbuffer. This is a DBG statement since any unexpected kill or
assert(stream);
/* Don't create anything if this is set for streaming. */
assert(stream);
/* Don't create anything if this is set for streaming. */
- if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) {
+ if (stream->relayd_id == (uint64_t) -1ULL && stream->chan->monitor) {
ret = utils_create_stream_file(stream->chan->pathname, stream->name,
stream->chan->tracefile_size, stream->tracefile_count_current,
stream->uid, stream->gid, NULL);
ret = utils_create_stream_file(stream->chan->pathname, stream->name,
stream->chan->tracefile_size, stream->tracefile_count_current,
stream->uid, stream->gid, NULL);