Fix: relayd trace file padding memleak

[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 f3d83451f94a559978b5d0dde88a133020da5916..c09c0c52582a45e03e5d7e292167bc907ff3bde6 100644 (file)
--- a/src/bin/lttng-relayd/main.c
+++ b/src/bin/lttng-relayd/main.c
@@ -1046,6 +1046,7 @@ int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr,
                goto end_unlock;
        }
 
+       stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num);
        stream->close_flag = 1;
 
        if (close_stream_check(stream)) {
@@ -1175,6 +1176,8 @@ static int write_padding_to_file(int fd, uint32_t size)
                PERROR("write padding to file");
        }
 
+       free(zeros);
+
 end:
        return ret;
 }
@@ -1207,9 +1210,12 @@ int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr,
        payload_size -= sizeof(struct lttcomm_relayd_metadata_payload);
 
        if (data_buffer_size < data_size) {
+               /* In case the realloc fails, we can free the memory */
+               char *tmp_data_ptr = data_buffer;
                data_buffer = realloc(data_buffer, data_size);
                if (!data_buffer) {
                        ERR("Allocating data buffer");
+                       free(tmp_data_ptr);
                        ret = -1;
                        goto end;
                }
@@ -1304,6 +1310,103 @@ end:
        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);
+
+       /* Avoid wrapping issue */
+       if (((int64_t) (stream->prev_seq - last_net_seq_num)) <= 0) {
+               /* 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
  */
@@ -1334,6 +1437,12 @@ int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr,
        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));
@@ -1378,9 +1487,11 @@ int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht)
 
        data_size = be32toh(data_hdr.data_size);
        if (data_buffer_size < data_size) {
+               char *tmp_data_ptr = data_buffer;
                data_buffer = realloc(data_buffer, data_size);
                if (!data_buffer) {
                        ERR("Allocating data buffer");
+                       free(tmp_data_ptr);
                        ret = -1;
                        goto end_unlock;
                }