Fix: consumer fd recv poll should be sensitive to POLLPRI too

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

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 3d79b5122d82bb3f020f3efe920d86b93c493005..2604f3d060574c1c7677047fcc16581a5c54818d 100644 (file)
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -2,24 +2,22 @@
  * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
  *                      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
  *
- * This program is free software; you can redistribute it and/or
- * modify it under the terms of the GNU General Public License
- * as published by the Free Software Foundation; only version 2
- * of the License.
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License, version 2 only,
+ * as published by the Free Software Foundation.
  *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU General Public License for more details.
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+ * more details.
  *
- * You should have received a copy of the GNU General Public License
- * along with this program; if not, write to the Free Software
- * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  */
 
 #define _GNU_SOURCE
 #include <assert.h>
-#include <fcntl.h>
 #include <poll.h>
 #include <pthread.h>
 #include <stdlib.h>
@@ -300,11 +298,23 @@ end:
 int consumer_add_stream(struct lttng_consumer_stream *stream)
 {
        int ret = 0;
+       struct lttng_ht_node_ulong *node;
+       struct lttng_ht_iter iter;
 
        pthread_mutex_lock(&consumer_data.lock);
        /* Steal stream identifier, for UST */
        consumer_steal_stream_key(stream->key);
        rcu_read_lock();
+
+       lttng_ht_lookup(consumer_data.stream_ht,
+                       (void *)((unsigned long) stream->key), &iter);
+       node = lttng_ht_iter_get_node_ulong(&iter);
+       if (node != NULL) {
+               rcu_read_unlock();
+               /* Stream already exist. Ignore the insertion */
+               goto end;
+       }
+
        lttng_ht_add_unique_ulong(consumer_data.stream_ht, &stream->node);
        rcu_read_unlock();
        consumer_data.stream_count++;
@@ -468,11 +478,25 @@ end:
  */
 int consumer_add_channel(struct lttng_consumer_channel *channel)
 {
+       struct lttng_ht_node_ulong *node;
+       struct lttng_ht_iter iter;
+
        pthread_mutex_lock(&consumer_data.lock);
        /* Steal channel identifier, for UST */
        consumer_steal_channel_key(channel->key);
        rcu_read_lock();
+
+       lttng_ht_lookup(consumer_data.channel_ht,
+                       (void *)((unsigned long) channel->key), &iter);
+       node = lttng_ht_iter_get_node_ulong(&iter);
+       if (node != NULL) {
+               /* Channel already exist. Ignore the insertion */
+               goto end;
+       }
+
        lttng_ht_add_unique_ulong(consumer_data.channel_ht, &channel->node);
+
+end:
        rcu_read_unlock();
        pthread_mutex_unlock(&consumer_data.lock);
 
@@ -514,7 +538,7 @@ int consumer_update_poll_array(
         * increment i so nb_fd is the number of real FD.
         */
        (*pollfd)[i].fd = ctx->consumer_poll_pipe[0];
-       (*pollfd)[i].events = POLLIN;
+       (*pollfd)[i].events = POLLIN | POLLPRI;
        return i;
 }
 
@@ -538,7 +562,7 @@ restart:
                perror("Poll error");
                goto exit;
        }
-       if (consumer_sockpoll[0].revents == POLLIN) {
+       if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) {
                DBG("consumer_should_quit wake up");
                goto exit;
        }
@@ -641,7 +665,7 @@ void lttng_consumer_sync_trace_file(
        if (orig_offset < stream->chan->max_sb_size) {
                return;
        }
-       sync_file_range(outfd, orig_offset - stream->chan->max_sb_size,
+       lttng_sync_file_range(outfd, orig_offset - stream->chan->max_sb_size,
                        stream->chan->max_sb_size,
                        SYNC_FILE_RANGE_WAIT_BEFORE
                        | SYNC_FILE_RANGE_WRITE
@@ -809,6 +833,8 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
                ERR("Unknown consumer_data type");
                assert(0);
        }
+
+       return 0;
 }
 
 /*
@@ -992,7 +1018,7 @@ void *lttng_consumer_thread_poll_fds(void *data)
                 * array. We want to prioritize array update over
                 * low-priority reads.
                 */
-               if (pollfd[nb_fd].revents & POLLIN) {
+               if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
                        DBG("consumer_poll_pipe wake up");
                        tmp2 = read(ctx->consumer_poll_pipe[0], &tmp, 1);
                        if (tmp2 < 0) {