Build fix: consumerd misnamed label
[lttng-tools.git] / src / common / consumer / consumer.c
index 2963c801d5407f14c3993552aa50cee75bcb1d51..91725d5d46be64f03935dc4a66f634d1ce5245c1 100644 (file)
@@ -1,22 +1,13 @@
 /*
- * Copyright (C) 2011 Julien Desfossez <julien.desfossez@polymtl.ca>
- *                      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
- *               2012 - David Goulet <dgoulet@efficios.com>
+ * Copyright (C) 2011 Julien Desfossez <julien.desfossez@polymtl.ca>
+ * Copyright (C) 2011 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ * Copyright (C) 2012 David Goulet <dgoulet@efficios.com>
  *
- * 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.
+ * SPDX-License-Identifier: GPL-2.0-only
  *
- * 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.,
- * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  */
 
+#include "common/index/ctf-index.h"
 #define _LGPL_SOURCE
 #include <assert.h>
 #include <poll.h>
@@ -578,95 +569,6 @@ void consumer_stream_update_channel_attributes(
                        channel->tracefile_size;
 }
 
-struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
-               uint64_t stream_key,
-               const char *channel_name,
-               uint64_t relayd_id,
-               uint64_t session_id,
-               struct lttng_trace_chunk *trace_chunk,
-               int cpu,
-               int *alloc_ret,
-               enum consumer_channel_type type,
-               unsigned int monitor)
-{
-       int ret;
-       struct lttng_consumer_stream *stream;
-
-       stream = zmalloc(sizeof(*stream));
-       if (stream == NULL) {
-               PERROR("malloc struct lttng_consumer_stream");
-               ret = -ENOMEM;
-               goto end;
-       }
-
-       if (trace_chunk && !lttng_trace_chunk_get(trace_chunk)) {
-               ERR("Failed to acquire trace chunk reference during the creation of a stream");
-               ret = -1;
-               goto error;
-       }
-
-       rcu_read_lock();
-       stream->key = stream_key;
-       stream->trace_chunk = trace_chunk;
-       stream->out_fd = -1;
-       stream->out_fd_offset = 0;
-       stream->output_written = 0;
-       stream->net_seq_idx = relayd_id;
-       stream->session_id = session_id;
-       stream->monitor = monitor;
-       stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
-       stream->index_file = NULL;
-       stream->last_sequence_number = -1ULL;
-       stream->rotate_position = -1ULL;
-       pthread_mutex_init(&stream->lock, NULL);
-       pthread_mutex_init(&stream->metadata_timer_lock, NULL);
-
-       /* If channel is the metadata, flag this stream as metadata. */
-       if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
-               stream->metadata_flag = 1;
-               /* Metadata is flat out. */
-               strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
-               /* Live rendez-vous point. */
-               pthread_cond_init(&stream->metadata_rdv, NULL);
-               pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
-       } else {
-               /* Format stream name to <channel_name>_<cpu_number> */
-               ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
-                               channel_name, cpu);
-               if (ret < 0) {
-                       PERROR("snprintf stream name");
-                       goto error;
-               }
-       }
-
-       /* Key is always the wait_fd for streams. */
-       lttng_ht_node_init_u64(&stream->node, stream->key);
-
-       /* Init node per channel id key */
-       lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
-
-       /* Init session id node with the stream session id */
-       lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
-
-       DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
-                       " relayd_id %" PRIu64 ", session_id %" PRIu64,
-                       stream->name, stream->key, channel_key,
-                       stream->net_seq_idx, stream->session_id);
-
-       rcu_read_unlock();
-       return stream;
-
-error:
-       rcu_read_unlock();
-       lttng_trace_chunk_put(stream->trace_chunk);
-       free(stream);
-end:
-       if (alloc_ret) {
-               *alloc_ret = ret;
-       }
-       return NULL;
-}
-
 /*
  * Add a stream to the global list protected by a mutex.
  */
@@ -1073,6 +975,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
                uint64_t session_id_per_pid,
                unsigned int monitor,
                unsigned int live_timer_interval,
+               bool is_in_live_session,
                const char *root_shm_path,
                const char *shm_path)
 {
@@ -1104,6 +1007,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
        channel->tracefile_count = tracefile_count;
        channel->monitor = monitor;
        channel->live_timer_interval = live_timer_interval;
+       channel->is_live = is_in_live_session;
        pthread_mutex_init(&channel->lock, NULL);
        pthread_mutex_init(&channel->timer_lock, NULL);
 
@@ -1472,7 +1376,7 @@ void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
 struct lttng_consumer_local_data *lttng_consumer_create(
                enum lttng_consumer_type type,
                ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
-                       struct lttng_consumer_local_data *ctx),
+                       struct lttng_consumer_local_data *ctx, bool locked_by_caller),
                int (*recv_channel)(struct lttng_consumer_channel *channel),
                int (*recv_stream)(struct lttng_consumer_stream *stream),
                int (*update_stream)(uint64_t stream_key, uint32_t state))
@@ -1679,19 +1583,18 @@ end:
  * Returns the number of bytes written
  */
 ssize_t lttng_consumer_on_read_subbuffer_mmap(
-               struct lttng_consumer_local_data *ctx,
-               struct lttng_consumer_stream *stream, unsigned long len,
-               unsigned long padding,
-               struct ctf_packet_index *index)
+               struct lttng_consumer_stream *stream,
+               const struct lttng_buffer_view *buffer,
+               unsigned long padding)
 {
-       unsigned long mmap_offset;
-       void *mmap_base;
        ssize_t ret = 0;
        off_t orig_offset = stream->out_fd_offset;
        /* Default is on the disk */
        int outfd = stream->out_fd;
        struct consumer_relayd_sock_pair *relayd = NULL;
        unsigned int relayd_hang_up = 0;
+       const size_t subbuf_content_size = buffer->size - padding;
+       size_t write_len;
 
        /* RCU lock for the relayd pointer */
        rcu_read_lock();
@@ -1707,39 +1610,9 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
                }
        }
 
-       /* get the offset inside the fd to mmap */
-       switch (consumer_data.type) {
-       case LTTNG_CONSUMER_KERNEL:
-               mmap_base = stream->mmap_base;
-               ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset);
-               if (ret < 0) {
-                       PERROR("tracer ctl get_mmap_read_offset");
-                       goto end;
-               }
-               break;
-       case LTTNG_CONSUMER32_UST:
-       case LTTNG_CONSUMER64_UST:
-               mmap_base = lttng_ustctl_get_mmap_base(stream);
-               if (!mmap_base) {
-                       ERR("read mmap get mmap base for stream %s", stream->name);
-                       ret = -EPERM;
-                       goto end;
-               }
-               ret = lttng_ustctl_get_mmap_read_offset(stream, &mmap_offset);
-               if (ret != 0) {
-                       PERROR("tracer ctl get_mmap_read_offset");
-                       ret = -EINVAL;
-                       goto end;
-               }
-               break;
-       default:
-               ERR("Unknown consumer_data type");
-               assert(0);
-       }
-
        /* Handle stream on the relayd if the output is on the network */
        if (relayd) {
-               unsigned long netlen = len;
+               unsigned long netlen = subbuf_content_size;
 
                /*
                 * Lock the control socket for the complete duration of the function
@@ -1777,10 +1650,10 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
                                goto write_error;
                        }
                }
-       } else {
-               /* No streaming, we have to set the len with the full padding */
-               len += padding;
 
+               write_len = subbuf_content_size;
+       } else {
+               /* No streaming; we have to write the full padding. */
                if (stream->metadata_flag && stream->reset_metadata_flag) {
                        ret = utils_truncate_stream_file(stream->out_fd, 0);
                        if (ret < 0) {
@@ -1794,7 +1667,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
                 * Check if we need to change the tracefile before writing the packet.
                 */
                if (stream->chan->tracefile_size > 0 &&
-                               (stream->tracefile_size_current + len) >
+                               (stream->tracefile_size_current + buffer->size) >
                                stream->chan->tracefile_size) {
                        ret = consumer_stream_rotate_output_files(stream);
                        if (ret) {
@@ -1803,19 +1676,17 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
                        outfd = stream->out_fd;
                        orig_offset = 0;
                }
-               stream->tracefile_size_current += len;
-               if (index) {
-                       index->offset = htobe64(stream->out_fd_offset);
-               }
+               stream->tracefile_size_current += buffer->size;
+               write_len = buffer->size;
        }
 
        /*
         * This call guarantee that len or less is returned. It's impossible to
         * receive a ret value that is bigger than len.
         */
-       ret = lttng_write(outfd, mmap_base + mmap_offset, len);
-       DBG("Consumer mmap write() ret %zd (len %lu)", ret, len);
-       if (ret < 0 || ((size_t) ret != len)) {
+       ret = lttng_write(outfd, buffer->data, write_len);
+       DBG("Consumer mmap write() ret %zd (len %zu)", ret, write_len);
+       if (ret < 0 || ((size_t) ret != write_len)) {
                /*
                 * Report error to caller if nothing was written else at least send the
                 * amount written.
@@ -1836,7 +1707,8 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
                        DBG("Consumer mmap write detected relayd hang up");
                } else {
                        /* Unhandled error, print it and stop function right now. */
-                       PERROR("Error in write mmap (ret %zd != len %lu)", ret, len);
+                       PERROR("Error in write mmap (ret %zd != write_len %zu)", ret,
+                                       write_len);
                }
                goto write_error;
        }
@@ -1845,9 +1717,9 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
        /* This call is useless on a socket so better save a syscall. */
        if (!relayd) {
                /* This won't block, but will start writeout asynchronously */
-               lttng_sync_file_range(outfd, stream->out_fd_offset, len,
+               lttng_sync_file_range(outfd, stream->out_fd_offset, write_len,
                                SYNC_FILE_RANGE_WRITE);
-               stream->out_fd_offset += len;
+               stream->out_fd_offset += write_len;
                lttng_consumer_sync_trace_file(stream, orig_offset);
        }
 
@@ -1881,8 +1753,7 @@ end:
 ssize_t lttng_consumer_on_read_subbuffer_splice(
                struct lttng_consumer_local_data *ctx,
                struct lttng_consumer_stream *stream, unsigned long len,
-               unsigned long padding,
-               struct ctf_packet_index *index)
+               unsigned long padding)
 {
        ssize_t ret = 0, written = 0, ret_splice = 0;
        loff_t offset = 0;
@@ -1986,7 +1857,6 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
                        orig_offset = 0;
                }
                stream->tracefile_size_current += len;
-               index->offset = htobe64(stream->out_fd_offset);
        }
 
        while (len > 0) {
@@ -2553,7 +2423,7 @@ restart:
                                do {
                                        health_code_update();
 
-                                       len = ctx->on_buffer_ready(stream, ctx);
+                                       len = ctx->on_buffer_ready(stream, ctx, false);
                                        /*
                                         * We don't check the return value here since if we get
                                         * a negative len, it means an error occurred thus we
@@ -2580,7 +2450,7 @@ restart:
                                        do {
                                                health_code_update();
 
-                                               len = ctx->on_buffer_ready(stream, ctx);
+                                               len = ctx->on_buffer_ready(stream, ctx, false);
                                                /*
                                                 * We don't check the return value here since if we get
                                                 * a negative len, it means an error occurred thus we
@@ -2796,7 +2666,7 @@ void *consumer_thread_data_poll(void *data)
                        if (pollfd[i].revents & POLLPRI) {
                                DBG("Urgent read on fd %d", pollfd[i].fd);
                                high_prio = 1;
-                               len = ctx->on_buffer_ready(local_stream[i], ctx);
+                               len = ctx->on_buffer_ready(local_stream[i], ctx, false);
                                /* it's ok to have an unavailable sub-buffer */
                                if (len < 0 && len != -EAGAIN && len != -ENODATA) {
                                        /* Clean the stream and free it. */
@@ -2827,7 +2697,7 @@ void *consumer_thread_data_poll(void *data)
                                        local_stream[i]->hangup_flush_done ||
                                        local_stream[i]->has_data) {
                                DBG("Normal read on fd %d", pollfd[i].fd);
-                               len = ctx->on_buffer_ready(local_stream[i], ctx);
+                               len = ctx->on_buffer_ready(local_stream[i], ctx, false);
                                /* it's ok to have an unavailable sub-buffer */
                                if (len < 0 && len != -EAGAIN && len != -ENODATA) {
                                        /* Clean the stream and free it. */
@@ -3441,39 +3311,123 @@ error_testpoint:
 }
 
 ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
-               struct lttng_consumer_local_data *ctx)
+               struct lttng_consumer_local_data *ctx,
+               bool locked_by_caller)
 {
-       ssize_t ret;
+       ssize_t ret, written_bytes;
+       int rotation_ret;
+       struct stream_subbuffer subbuffer = {};
 
-       pthread_mutex_lock(&stream->chan->lock);
-       pthread_mutex_lock(&stream->lock);
-       if (stream->metadata_flag) {
-               pthread_mutex_lock(&stream->metadata_rdv_lock);
+       if (!locked_by_caller) {
+               stream->read_subbuffer_ops.lock(stream);
        }
 
-       switch (consumer_data.type) {
-       case LTTNG_CONSUMER_KERNEL:
-               ret = lttng_kconsumer_read_subbuffer(stream, ctx);
-               break;
-       case LTTNG_CONSUMER32_UST:
-       case LTTNG_CONSUMER64_UST:
-               ret = lttng_ustconsumer_read_subbuffer(stream, ctx);
-               break;
-       default:
-               ERR("Unknown consumer_data type");
-               assert(0);
-               ret = -ENOSYS;
-               break;
+       if (stream->read_subbuffer_ops.on_wake_up) {
+               ret = stream->read_subbuffer_ops.on_wake_up(stream);
+               if (ret) {
+                       goto end;
+               }
        }
 
-       if (stream->metadata_flag) {
-               pthread_cond_broadcast(&stream->metadata_rdv);
-               pthread_mutex_unlock(&stream->metadata_rdv_lock);
+       /*
+        * If the stream was flagged to be ready for rotation before we extract
+        * the next packet, rotate it now.
+        */
+       if (stream->rotate_ready) {
+               DBG("Rotate stream before consuming data");
+               ret = lttng_consumer_rotate_stream(ctx, stream);
+               if (ret < 0) {
+                       ERR("Stream rotation error before consuming data");
+                       goto end;
+               }
+       }
+
+       ret = stream->read_subbuffer_ops.get_next_subbuffer(stream, &subbuffer);
+       if (ret) {
+               if (ret == -ENODATA) {
+                       /* Not an error. */
+                       ret = 0;
+                       goto sleep_stream;
+               }
+               goto end;
+       }
+
+       ret = stream->read_subbuffer_ops.pre_consume_subbuffer(
+                       stream, &subbuffer);
+       if (ret) {
+               goto error_put_subbuf;
+       }
+
+       written_bytes = stream->read_subbuffer_ops.consume_subbuffer(
+                       ctx, stream, &subbuffer);
+       /*
+        * Should write subbuf_size amount of data when network streaming or
+        * the full padded size when we are not streaming.
+        */
+       if ((written_bytes != subbuffer.info.data.subbuf_size &&
+                           stream->net_seq_idx != (uint64_t) -1ULL) ||
+                       (written_bytes != subbuffer.info.data.padded_subbuf_size &&
+                                       stream->net_seq_idx ==
+                                                       (uint64_t) -1ULL)) {
+               /*
+                * Display the error but continue processing to try to
+                * release the subbuffer. This is a DBG statement
+                * since this can happen without being a critical
+                * error.
+                */
+               DBG("Failed to write to tracefile (written_bytes: %zd != padded subbuffer size: %lu, subbuffer size: %lu)",
+                               written_bytes, subbuffer.info.data.subbuf_size,
+                               subbuffer.info.data.padded_subbuf_size);
+       }
+
+       ret = stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer);
+       if (ret) {
+               goto end;
+       }
+
+       if (stream->read_subbuffer_ops.post_consume) {
+               ret = stream->read_subbuffer_ops.post_consume(stream, &subbuffer, ctx);
+               if (ret) {
+                       goto end;
+               }
+       }
+
+       /*
+        * After extracting the packet, we check if the stream is now ready to
+        * be rotated and perform the action immediately.
+        *
+        * Don't overwrite `ret` as callers expect the number of bytes
+        * consumed to be returned on success.
+        */
+       rotation_ret = lttng_consumer_stream_is_rotate_ready(stream);
+       if (rotation_ret == 1) {
+               rotation_ret = lttng_consumer_rotate_stream(ctx, stream);
+               if (rotation_ret < 0) {
+                       ret = rotation_ret;
+                       ERR("Stream rotation error after consuming data");
+                       goto end;
+               }
+       } else if (rotation_ret < 0) {
+               ret = rotation_ret;
+               ERR("Failed to check if stream was ready to rotate after consuming data");
+               goto end;
+       }
+
+sleep_stream:
+       if (stream->read_subbuffer_ops.on_sleep) {
+               stream->read_subbuffer_ops.on_sleep(stream, ctx);
+       }
+
+       ret = written_bytes;
+end:
+       if (!locked_by_caller) {
+               stream->read_subbuffer_ops.unlock(stream);
        }
-       pthread_mutex_unlock(&stream->lock);
-       pthread_mutex_unlock(&stream->chan->lock);
 
        return ret;
+error_put_subbuf:
+       (void) stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer);
+       goto end;
 }
 
 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
@@ -3963,8 +3917,18 @@ int consumer_flush_buffer(struct lttng_consumer_stream *stream, int producer_act
                } else {
                        ret = kernctl_buffer_flush_empty(stream->wait_fd);
                        if (ret < 0) {
-                               ERR("Failed to flush kernel stream");
-                               goto end;
+                               /*
+                                * Doing a buffer flush which does not take into
+                                * account empty packets. This is not perfect,
+                                * but required as a fall-back when
+                                * "flush_empty" is not implemented by
+                                * lttng-modules.
+                                */
+                               ret = kernctl_buffer_flush(stream->wait_fd);
+                               if (ret < 0) {
+                                       ERR("Failed to flush kernel stream");
+                                       goto end;
+                               }
                        }
                }
                break;
@@ -4084,7 +4048,13 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
                 */
                produced_pos = ALIGN_FLOOR(produced_pos, stream->max_sb_size);
                if (consumed_pos == produced_pos) {
+                       DBG("Set rotate ready for stream %" PRIu64 " produced = %lu consumed = %lu",
+                                       stream->key, produced_pos, consumed_pos);
                        stream->rotate_ready = true;
+               } else {
+                       DBG("Different consumed and produced positions "
+                                       "for stream %" PRIu64 " produced = %lu consumed = %lu",
+                                       stream->key, produced_pos, consumed_pos);
                }
                /*
                 * The rotation position is based on the packet_seq_num of the
@@ -4109,6 +4079,8 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
                }
                stream->rotate_position = stream->last_sequence_number + 1 +
                                ((produced_pos - consumed_pos) / stream->max_sb_size);
+               DBG("Set rotation position for stream %" PRIu64 " at position %" PRIu64,
+                               stream->key, stream->rotate_position);
 
                if (!is_local_trace) {
                        /*
@@ -4173,6 +4145,110 @@ end:
        return ret;
 }
 
+static
+int consumer_clear_buffer(struct lttng_consumer_stream *stream)
+{
+       int ret = 0;
+       unsigned long consumed_pos_before, consumed_pos_after;
+
+       ret = lttng_consumer_sample_snapshot_positions(stream);
+       if (ret < 0) {
+               ERR("Taking snapshot positions");
+               goto end;
+       }
+
+       ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos_before);
+       if (ret < 0) {
+               ERR("Consumed snapshot position");
+               goto end;
+       }
+
+       switch (consumer_data.type) {
+       case LTTNG_CONSUMER_KERNEL:
+               ret = kernctl_buffer_clear(stream->wait_fd);
+               if (ret < 0) {
+                       ERR("Failed to clear kernel stream (ret = %d)", ret);
+                       goto end;
+               }
+               break;
+       case LTTNG_CONSUMER32_UST:
+       case LTTNG_CONSUMER64_UST:
+               lttng_ustconsumer_clear_buffer(stream);
+               break;
+       default:
+               ERR("Unknown consumer_data type");
+               abort();
+       }
+
+       ret = lttng_consumer_sample_snapshot_positions(stream);
+       if (ret < 0) {
+               ERR("Taking snapshot positions");
+               goto end;
+       }
+       ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos_after);
+       if (ret < 0) {
+               ERR("Consumed snapshot position");
+               goto end;
+       }
+       DBG("clear: before: %lu after: %lu", consumed_pos_before, consumed_pos_after);
+end:
+       return ret;
+}
+
+static
+int consumer_clear_stream(struct lttng_consumer_stream *stream)
+{
+       int ret;
+
+       ret = consumer_flush_buffer(stream, 1);
+       if (ret < 0) {
+               ERR("Failed to flush stream %" PRIu64 " during channel clear",
+                               stream->key);
+               ret = LTTCOMM_CONSUMERD_FATAL;
+               goto error;
+       }
+
+       ret = consumer_clear_buffer(stream);
+       if (ret < 0) {
+               ERR("Failed to clear stream %" PRIu64 " during channel clear",
+                               stream->key);
+               ret = LTTCOMM_CONSUMERD_FATAL;
+               goto error;
+       }
+
+       ret = LTTCOMM_CONSUMERD_SUCCESS;
+error:
+       return ret;
+}
+
+static
+int consumer_clear_unmonitored_channel(struct lttng_consumer_channel *channel)
+{
+       int ret;
+       struct lttng_consumer_stream *stream;
+
+       rcu_read_lock();
+       pthread_mutex_lock(&channel->lock);
+       cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
+               health_code_update();
+               pthread_mutex_lock(&stream->lock);
+               ret = consumer_clear_stream(stream);
+               if (ret) {
+                       goto error_unlock;
+               }
+               pthread_mutex_unlock(&stream->lock);
+       }
+       pthread_mutex_unlock(&channel->lock);
+       rcu_read_unlock();
+       return 0;
+
+error_unlock:
+       pthread_mutex_unlock(&stream->lock);
+       pthread_mutex_unlock(&channel->lock);
+       rcu_read_unlock();
+       return ret;
+}
+
 /*
  * Check if a stream is ready to be rotated after extracting it.
  *
@@ -4181,6 +4257,11 @@ end:
  */
 int lttng_consumer_stream_is_rotate_ready(struct lttng_consumer_stream *stream)
 {
+       DBG("Check is rotate ready for stream %" PRIu64
+                       " ready %u rotate_position %" PRIu64
+                       " last_sequence_number %" PRIu64,
+                       stream->key, stream->rotate_ready,
+                       stream->rotate_position, stream->last_sequence_number);
        if (stream->rotate_ready) {
                return 1;
        }
@@ -4209,6 +4290,12 @@ int lttng_consumer_stream_is_rotate_ready(struct lttng_consumer_stream *stream)
         * but consumerd considers rotation ready when reaching the last
         * packet of the current chunk, hence the "rotate_position - 1".
         */
+
+       DBG("Check is rotate ready for stream %" PRIu64
+                       " last_sequence_number %" PRIu64
+                       " rotate_position %" PRIu64,
+                       stream->key, stream->last_sequence_number,
+                       stream->rotate_position);
        if (stream->last_sequence_number >= stream->rotate_position - 1) {
                return 1;
        }
@@ -4221,6 +4308,8 @@ int lttng_consumer_stream_is_rotate_ready(struct lttng_consumer_stream *stream)
  */
 void lttng_consumer_reset_stream_rotate_state(struct lttng_consumer_stream *stream)
 {
+       DBG("lttng_consumer_reset_stream_rotate_state for stream %" PRIu64,
+                       stream->key);
        stream->rotate_position = -1ULL;
        stream->rotate_ready = false;
 }
@@ -4473,7 +4562,7 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk(
         * and LTTNG_CONSUMER_DESTROY_TRACE_CHUNK commands.
         */
        created_chunk = lttng_trace_chunk_create(chunk_id,
-                       chunk_creation_timestamp);
+                       chunk_creation_timestamp, NULL);
        if (!created_chunk) {
                ERR("Failed to create trace chunk");
                ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
@@ -4804,3 +4893,67 @@ end_rcu_unlock:
 end:
        return ret_code;
 }
+
+static
+int consumer_clear_monitored_channel(struct lttng_consumer_channel *channel)
+{
+       struct lttng_ht *ht;
+       struct lttng_consumer_stream *stream;
+       struct lttng_ht_iter iter;
+       int ret;
+
+       ht = consumer_data.stream_per_chan_id_ht;
+
+       rcu_read_lock();
+       cds_lfht_for_each_entry_duplicate(ht->ht,
+                       ht->hash_fct(&channel->key, lttng_ht_seed),
+                       ht->match_fct, &channel->key,
+                       &iter.iter, stream, node_channel_id.node) {
+               /*
+                * Protect against teardown with mutex.
+                */
+               pthread_mutex_lock(&stream->lock);
+               if (cds_lfht_is_node_deleted(&stream->node.node)) {
+                       goto next;
+               }
+               ret = consumer_clear_stream(stream);
+               if (ret) {
+                       goto error_unlock;
+               }
+       next:
+               pthread_mutex_unlock(&stream->lock);
+       }
+       rcu_read_unlock();
+       return LTTCOMM_CONSUMERD_SUCCESS;
+
+error_unlock:
+       pthread_mutex_unlock(&stream->lock);
+       rcu_read_unlock();
+       return ret;
+}
+
+int lttng_consumer_clear_channel(struct lttng_consumer_channel *channel)
+{
+       int ret;
+
+       DBG("Consumer clear channel %" PRIu64, channel->key);
+
+       if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA) {
+               /*
+                * Nothing to do for the metadata channel/stream.
+                * Snapshot mechanism already take care of the metadata
+                * handling/generation, and monitored channels only need to
+                * have their data stream cleared..
+                */
+               ret = LTTCOMM_CONSUMERD_SUCCESS;
+               goto end;
+       }
+
+       if (!channel->monitor) {
+               ret = consumer_clear_unmonitored_channel(channel);
+       } else {
+               ret = consumer_clear_monitored_channel(channel);
+       }
+end:
+       return ret;
+}
This page took 0.032041 seconds and 4 git commands to generate.