Add mkdir_recursive function to libcommon utils

[lttng-tools.git] / src / common / consumer.c

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 935a03d235fee67a32f1f6bd14fa7c9bdf65845d..08592f6c0b8713f1ce1765a7de4d9237fd8bd81a 100644 (file)
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -755,7 +755,7 @@ static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
                 * 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++);
+               data_hdr.net_seq_num = htobe64(stream->next_net_seq_num);
                /* Other fields are zeroed previously */
 
                ret = relayd_send_data_hdr(&relayd->data_sock, &data_hdr,
@@ -764,6 +764,8 @@ static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
                        goto error;
                }
 
+               ++stream->next_net_seq_num;
+
                /* Set to go on data socket */
                outfd = relayd->data_sock.fd;
        }
@@ -2281,14 +2283,11 @@ void *consumer_thread_data_poll(void *data)
                 */
                pthread_mutex_lock(&consumer_data.lock);
                if (consumer_data.need_update) {
-                       if (pollfd != NULL) {
-                               free(pollfd);
-                               pollfd = NULL;
-                       }
-                       if (local_stream != NULL) {
-                               free(local_stream);
-                               local_stream = NULL;
-                       }
+                       free(pollfd);
+                       pollfd = NULL;
+
+                       free(local_stream);
+                       local_stream = NULL;
 
                        /* allocate for all fds + 1 for the consumer_data_pipe */
                        pollfd = zmalloc((consumer_data.stream_count + 1) * sizeof(struct pollfd));
@@ -2349,7 +2348,7 @@ void *consumer_thread_data_poll(void *data)
                 * array update over low-priority reads.
                 */
                if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
-                       size_t pipe_readlen;
+                       ssize_t pipe_readlen;
 
                        DBG("consumer_data_pipe wake up");
                        /* Consume 1 byte of pipe data */
@@ -2485,14 +2484,8 @@ void *consumer_thread_data_poll(void *data)
        }
 end:
        DBG("polling thread exiting");
-       if (pollfd != NULL) {
-               free(pollfd);
-               pollfd = NULL;
-       }
-       if (local_stream != NULL) {
-               free(local_stream);
-               local_stream = NULL;
-       }
+       free(pollfd);
+       free(local_stream);
 
        /*
         * Close the write side of the pipe so epoll_wait() in