struct lttng_poll_event events;
struct lttcomm_sock *control_sock, *data_sock;
- /*
- * Get allocated in this thread, enqueued to a global queue, dequeued and
- * freed in the worker thread.
- */
- struct relay_command *relay_cmd = NULL;
-
DBG("[thread] Relay listener started");
control_sock = relay_init_sock(control_uri);
ERR("socket poll error");
goto error;
} else if (revents & LPOLLIN) {
- struct lttcomm_sock *newsock = NULL;
+ /*
+ * Get allocated in this thread,
+ * enqueued to a global queue, dequeued
+ * and freed in the worker thread.
+ */
+ struct relay_command *relay_cmd;
+ struct lttcomm_sock *newsock;
relay_cmd = zmalloc(sizeof(struct relay_command));
if (relay_cmd == NULL) {
if (pollfd == data_sock->fd) {
newsock = data_sock->ops->accept(data_sock);
- if (newsock < 0) {
+ if (!newsock) {
PERROR("accepting data sock");
+ free(relay_cmd);
goto error;
}
relay_cmd->type = RELAY_DATA;
DBG("Relay data connection accepted, socket %d", newsock->fd);
- } else if (pollfd == control_sock->fd) {
+ } else {
+ assert(pollfd == control_sock->fd);
newsock = control_sock->ops->accept(control_sock);
- if (newsock < 0) {
+ if (!newsock) {
PERROR("accepting control sock");
+ free(relay_cmd);
goto error;
}
relay_cmd->type = RELAY_CONTROL;
&val, sizeof(int));
if (ret < 0) {
PERROR("setsockopt inet");
+ lttcomm_destroy_sock(newsock);
+ free(relay_cmd);
goto error;
}
relay_cmd->sock = newsock;
goto end_unlock;
}
+ stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num);
stream->close_flag = 1;
if (close_stream_check(stream)) {
int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
struct relay_command *cmd)
{
- int ret = htobe32(LTTNG_OK);
+ int ret;
struct lttcomm_relayd_version reply;
- struct relay_session *session = NULL;
+ struct relay_session *session;
if (cmd->session == NULL) {
session = zmalloc(sizeof(struct relay_session));
session->id = ++last_relay_session_id;
DBG("Created session %" PRIu64, session->id);
cmd->session = session;
+ } else {
+ session = cmd->session;
}
session->version_check_done = 1;
- sscanf(VERSION, "%u.%u", &reply.major, &reply.minor);
+ ret = sscanf(VERSION, "%u.%u", &reply.major, &reply.minor);
+ if (ret < 2) {
+ ERR("Error in scanning version");
+ ret = -1;
+ goto end;
+ }
reply.major = htobe32(reply.major);
reply.minor = htobe32(reply.minor);
ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
return ret;
}
+/*
+ * Check for data availability for a given stream id from the session daemon.
+ */
+static
+int relay_data_available(struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_command *cmd, struct lttng_ht *streams_ht)
+{
+ struct relay_session *session = cmd->session;
+ struct lttcomm_relayd_data_available msg;
+ struct lttcomm_relayd_generic_reply reply;
+ struct relay_stream *stream;
+ int ret;
+ struct lttng_ht_node_ulong *node;
+ struct lttng_ht_iter iter;
+ uint64_t last_net_seq_num, stream_id;
+
+ DBG("Data available command received");
+
+ if (!session || session->version_check_done == 0) {
+ ERR("Trying to check for data before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), MSG_WAITALL);
+ if (ret < sizeof(msg)) {
+ ERR("Relay didn't receive valid data_available struct size : %d", ret);
+ ret = -1;
+ goto end_no_session;
+ }
+
+ stream_id = be64toh(msg.stream_id);
+ last_net_seq_num = be64toh(msg.last_net_seq_num);
+
+ rcu_read_lock();
+ lttng_ht_lookup(streams_ht, (void *)((unsigned long) stream_id), &iter);
+ node = lttng_ht_iter_get_node_ulong(&iter);
+ if (node == NULL) {
+ DBG("Relay stream %" PRIu64 " not found", stream_id);
+ ret = -1;
+ goto end_unlock;
+ }
+
+ stream = caa_container_of(node, struct relay_stream, stream_n);
+ assert(stream);
+
+ DBG("Data available for stream id %" PRIu64 " prev_seq %" PRIu64
+ " and last_seq %" PRIu64, stream_id, stream->prev_seq,
+ last_net_seq_num);
+
+ if (stream->prev_seq == -1UL || stream->prev_seq <= last_net_seq_num) {
+ /* Data has in fact been written and is available */
+ ret = 1;
+ } else {
+ /* Data still being streamed. */
+ ret = 0;
+ }
+
+end_unlock:
+ rcu_read_unlock();
+
+ reply.ret_code = htobe32(ret);
+ ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
+ if (ret < 0) {
+ ERR("Relay data available ret code failed");
+ }
+
+end_no_session:
+ return ret;
+}
+
+/*
+ * Wait for the control socket to reach a quiescent state.
+ *
+ * Note that for now, when receiving this command from the session daemon, this
+ * means that every subsequent commands or data received on the control socket
+ * has been handled. So, this is why we simply return OK here.
+ */
+static
+int relay_quiescent_control(struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_command *cmd)
+{
+ int ret;
+ struct lttcomm_relayd_generic_reply reply;
+
+ DBG("Checking quiescent state on control socket");
+
+ reply.ret_code = htobe32(LTTNG_OK);
+ ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
+ if (ret < 0) {
+ ERR("Relay data available ret code failed");
+ }
+
+ return ret;
+}
+
/*
* relay_process_control: Process the commands received on the control socket
*/
case RELAYD_CLOSE_STREAM:
ret = relay_close_stream(recv_hdr, cmd, streams_ht);
break;
+ case RELAYD_DATA_AVAILABLE:
+ ret = relay_data_available(recv_hdr, cmd, streams_ht);
+ break;
+ case RELAYD_QUIESCENT_CONTROL:
+ ret = relay_quiescent_control(recv_hdr, cmd);
+ break;
case RELAYD_UPDATE_SYNC_INFO:
default:
ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
goto end_unlock;
}
+ DBG2("Relay wrote %d bytes to tracefile for stream id %" PRIu64,
+ ret, stream->stream_handle);
+
ret = write_padding_to_file(stream->fd, be32toh(data_hdr.padding_size));
if (ret < 0) {
goto end_unlock;
}
- DBG2("Relay wrote %d bytes to tracefile for stream id %" PRIu64,
- ret, stream->stream_handle);
-
stream->prev_seq = net_seq_num;
/* Check if we need to close the FD */
projects / lttng-tools.git / blobdiff