data_hdr.stream_id = htobe64(stream->relayd_stream_id);
data_hdr.data_size = htobe32(data_size);
data_hdr.padding_size = htobe32(padding);
+ /*
+ * Note that net_seq_num below is assigned with the *current* value of
+ * next_net_seq_num and only after that the next_net_seq_num will be
+ * increment. This is why when issuing a command on the relayd using
+ * this next value, 1 should always be substracted in order to compare
+ * the last seen sequence number on the relayd side to the last sent.
+ */
data_hdr.net_seq_num = htobe64(stream->next_net_seq_num++);
/* Other fields are zeroed previously */
do {
ret = write(ctx->consumer_should_quit[1], "4", 1);
} while (ret < 0 && errno == EINTR);
- if (ret < 0) {
+ if (ret < 0 || ret != 1) {
PERROR("write consumer quit");
}
do {
ret = write(fd, (void *) &hdr, sizeof(hdr));
} while (ret < 0 && errno == EINTR);
- if (ret < 0) {
- PERROR("write metadata stream id");
+ if (ret < 0 || ret != sizeof(hdr)) {
+ /*
+ * This error means that the fd's end is closed so ignore the perror
+ * not to clubber the error output since this can happen in a normal
+ * code path.
+ */
+ if (errno != EPIPE) {
+ PERROR("write metadata stream id");
+ }
+ DBG3("Consumer failed to write relayd metadata id (errno: %d)", errno);
+ /*
+ * Set ret to a negative value because if ret != sizeof(hdr), we don't
+ * handle writting the missing part so report that as an error and
+ * don't lie to the caller.
+ */
+ ret = -1;
goto end;
}
DBG("Metadata stream id %" PRIu64 " with padding %lu written before data",
} while (ret < 0 && errno == EINTR);
DBG("Consumer mmap write() ret %zd (len %lu)", ret, len);
if (ret < 0) {
- PERROR("Error in file write");
+ /*
+ * This is possible if the fd is closed on the other side (outfd)
+ * or any write problem. It can be verbose a bit for a normal
+ * execution if for instance the relayd is stopped abruptly. This
+ * can happen so set this to a DBG statement.
+ */
+ DBG("Error in file write mmap");
if (written == 0) {
written = ret;
}
DBG("Metadata main loop started");
while (1) {
- lttng_poll_reset(&events);
-
- nb_fd = LTTNG_POLL_GETNB(&events);
-
/* Only the metadata pipe is set */
- if (nb_fd == 0 && consumer_quit == 1) {
+ if (LTTNG_POLL_GETNB(&events) == 0 && consumer_quit == 1) {
goto end;
}
restart:
- DBG("Metadata poll wait with %d fd(s)", nb_fd);
+ DBG("Metadata poll wait with %d fd(s)", LTTNG_POLL_GETNB(&events));
ret = lttng_poll_wait(&events, -1);
DBG("Metadata event catched in thread");
if (ret < 0) {
goto error;
}
+ nb_fd = ret;
+
/* From here, the event is a metadata wait fd */
for (i = 0; i < nb_fd; i++) {
revents = LTTNG_POLL_GETEV(&events, i);
pipe_readlen = read(ctx->consumer_data_pipe[0], &new_stream,
sizeof(new_stream));
} while (pipe_readlen == -1 && errno == EINTR);
+ if (pipe_readlen < 0) {
+ PERROR("read consumer data pipe");
+ /* Continue so we can at least handle the current stream(s). */
+ continue;
+ }
/*
* If the stream is NULL, just ignore it. It's also possible that
/* Blocking call, waiting for transmission */
sock = lttcomm_accept_unix_sock(client_socket);
- if (sock <= 0) {
+ if (sock < 0) {
WARN("On accept");
goto end;
}
relayd = consumer_allocate_relayd_sock_pair(net_seq_idx);
if (relayd == NULL) {
lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
+ ret = -1;
goto error;
}
relayd->sessiond_session_id = (uint64_t) sessiond_id;
relayd->control_sock.fd = fd;
/*
- * Create a session on the relayd and store the returned id. No need to
- * grab the socket lock since the relayd object is not yet visible.
+ * Create a session on the relayd and store the returned id. Lock the
+ * control socket mutex if the relayd was NOT created before.
*/
+ if (!relayd_created) {
+ pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+ }
ret = relayd_create_session(&relayd->control_sock,
&relayd->relayd_session_id);
+ if (!relayd_created) {
+ pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+ }
if (ret < 0) {
goto error;
}
relayd_id_node = zmalloc(sizeof(struct consumer_relayd_session_id));
if (!relayd_id_node) {
PERROR("zmalloc relayd id node");
+ ret = -1;
goto error;
}
break;
default:
ERR("Unknown relayd socket type (%d)", sock_type);
+ ret = -1;
goto error;
}
static struct consumer_relayd_sock_pair *find_relayd_by_session_id(uint64_t id)
{
struct lttng_ht_iter iter;
- struct lttng_ht_node_ulong *node;
struct consumer_relayd_sock_pair *relayd = NULL;
- struct consumer_relayd_session_id *session_id_map;
-
- /* Get the session id map. */
- lttng_ht_lookup(relayd_session_id_ht, (void *)((unsigned long) id), &iter);
- node = lttng_ht_iter_get_node_ulong(&iter);
- if (node == NULL) {
- goto end;
- }
-
- session_id_map = caa_container_of(node, struct consumer_relayd_session_id,
- node);
/* Iterate over all relayd since they are indexed by net_seq_idx. */
cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd,
node.node) {
- if (relayd->relayd_session_id == session_id_map->relayd_id) {
+ /*
+ * Check by sessiond id which is unique here where the relayd session
+ * id might not be when having multiple relayd.
+ */
+ if (relayd->sessiond_session_id == id) {
/* Found the relayd. There can be only one per id. */
- break;
+ goto found;
}
}
-end:
+ return NULL;
+
+found:
return relayd;
}
if (relayd) {
pthread_mutex_lock(&relayd->ctrl_sock_mutex);
if (stream->metadata_flag) {
- ret = relayd_quiescent_control(&relayd->control_sock);
+ ret = relayd_quiescent_control(&relayd->control_sock,
+ stream->relayd_stream_id);
} else {
ret = relayd_data_pending(&relayd->control_sock,
- stream->relayd_stream_id, stream->next_net_seq_num);
+ stream->relayd_stream_id,
+ stream->next_net_seq_num - 1);
}
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret == 1) {
ret = relayd_end_data_pending(&relayd->control_sock,
relayd->relayd_session_id, &is_data_inflight);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
- if (ret < 0 || !is_data_inflight) {
- /* On error or if NO data inflight, no data is pending. */
+ if (ret < 0) {
goto data_not_pending;
}
+ if (is_data_inflight) {
+ goto data_pending;
+ }
}
/*
projects / lttng-tools.git / blobdiff