src/common/consumer/consumer-metadata-cache.cpp

   1 /*
   2  * Copyright (C) 2013 Julien Desfossez <jdesfossez@efficios.com>
   3  * Copyright (C) 2013 David Goulet <dgoulet@efficios.com>
   4  *
   5  * SPDX-License-Identifier: GPL-2.0-only
   6  *
   7  */
   8
   9 #define _LGPL_SOURCE
  10 #include <pthread.h>
  11 #include <stdlib.h>
  12 #include <string.h>
  13 #include <sys/types.h>
  14 #include <unistd.h>
  15 #include <inttypes.h>
  16
  17 #include <common/common.hpp>
  18 #include <common/utils.hpp>
  19 #include <common/sessiond-comm/sessiond-comm.hpp>
  20 #include <common/ust-consumer/ust-consumer.hpp>
  21 #include <common/consumer/consumer.hpp>
  22
  23 #include "consumer-metadata-cache.hpp"
  24
  25 enum metadata_cache_update_version_status {
  26         METADATA_CACHE_UPDATE_STATUS_VERSION_UPDATED,
  27         METADATA_CACHE_UPDATE_STATUS_VERSION_NOT_UPDATED,
  28 };
  29
  30 extern struct lttng_consumer_global_data the_consumer_data;
  31
  32 /*
  33  * Reset the metadata cache.
  34  */
  35 static
  36 void metadata_cache_reset(struct consumer_metadata_cache *cache)
  37 {
  38         const int ret = lttng_dynamic_buffer_set_size(&cache->contents, 0);
  39
  40         LTTNG_ASSERT(ret == 0);
  41 }
  42
  43 /*
  44  * Check if the metadata cache version changed.
  45  * If it did, reset the metadata cache.
  46  * The metadata cache lock MUST be held.
  47  */
  48 static enum metadata_cache_update_version_status metadata_cache_update_version(
  49                 struct consumer_metadata_cache *cache, uint64_t version)
  50 {
  51         enum metadata_cache_update_version_status status;
  52
  53         if (cache->version == version) {
  54                 status = METADATA_CACHE_UPDATE_STATUS_VERSION_NOT_UPDATED;
  55                 goto end;
  56         }
  57
  58         DBG("Metadata cache version update to %" PRIu64, version);
  59         cache->version = version;
  60         status = METADATA_CACHE_UPDATE_STATUS_VERSION_UPDATED;
  61
  62 end:
  63         return status;
  64 }
  65
  66 /*
  67  * Write metadata to the cache, extend the cache if necessary. We support
  68  * overlapping updates, but they need to be contiguous. Send the
  69  * contiguous metadata in cache to the ring buffer. The metadata cache
  70  * lock MUST be acquired to write in the cache.
  71  *
  72  * See `enum consumer_metadata_cache_write_status` for the meaning of the
  73  * various returned status codes.
  74  */
  75 enum consumer_metadata_cache_write_status
  76 consumer_metadata_cache_write(struct consumer_metadata_cache *cache,
  77                 unsigned int offset, unsigned int len, uint64_t version,
  78                 const char *data)
  79 {
  80         int ret = 0;
  81         enum consumer_metadata_cache_write_status status;
  82         bool cache_is_invalidated = false;
  83         uint64_t original_size;
  84
  85         LTTNG_ASSERT(cache);
  86         ASSERT_LOCKED(cache->lock);
  87         original_size = cache->contents.size;
  88
  89         if (metadata_cache_update_version(cache, version) ==
  90                         METADATA_CACHE_UPDATE_STATUS_VERSION_UPDATED) {
  91                 metadata_cache_reset(cache);
  92                 cache_is_invalidated = true;
  93         }
  94
  95         DBG("Writing %u bytes from offset %u in metadata cache", len, offset);
  96         if (offset + len > cache->contents.size) {
  97                 ret = lttng_dynamic_buffer_set_size(
  98                                 &cache->contents, offset + len);
  99                 if (ret) {
 100                         ERR("Extending metadata cache");
 101                         status = CONSUMER_METADATA_CACHE_WRITE_STATUS_ERROR;
 102                         goto end;
 103                 }
 104         }
 105
 106         memcpy(cache->contents.data + offset, data, len);
 107
 108         if (cache_is_invalidated) {
 109                 status = CONSUMER_METADATA_CACHE_WRITE_STATUS_INVALIDATED;
 110         } else if (cache->contents.size > original_size) {
 111                 status = CONSUMER_METADATA_CACHE_WRITE_STATUS_APPENDED_CONTENT;
 112         } else {
 113                 status = CONSUMER_METADATA_CACHE_WRITE_STATUS_NO_CHANGE;
 114                 LTTNG_ASSERT(cache->contents.size == original_size);
 115         }
 116
 117 end:
 118         return status;
 119 }
 120
 121 /*
 122  * Create the metadata cache, original allocated size: max_sb_size
 123  *
 124  * Return 0 on success, a negative value on error.
 125  */
 126 int consumer_metadata_cache_allocate(struct lttng_consumer_channel *channel)
 127 {
 128         int ret;
 129
 130         LTTNG_ASSERT(channel);
 131
 132         channel->metadata_cache = zmalloc<consumer_metadata_cache>();
 133         if (!channel->metadata_cache) {
 134                 PERROR("zmalloc metadata cache struct");
 135                 ret = -1;
 136                 goto end;
 137         }
 138         ret = pthread_mutex_init(&channel->metadata_cache->lock, NULL);
 139         if (ret != 0) {
 140                 PERROR("mutex init");
 141                 goto end_free_cache;
 142         }
 143
 144         lttng_dynamic_buffer_init(&channel->metadata_cache->contents);
 145         ret = lttng_dynamic_buffer_set_capacity(
 146                         &channel->metadata_cache->contents,
 147                         DEFAULT_METADATA_CACHE_SIZE);
 148         if (ret) {
 149                 PERROR("Failed to pre-allocate metadata cache storage of %d bytes on creation",
 150                                 DEFAULT_METADATA_CACHE_SIZE);
 151                 ret = -1;
 152                 goto end_free_mutex;
 153         }
 154
 155         DBG("Allocated metadata cache: current capacity = %zu",
 156                         lttng_dynamic_buffer_get_capacity_left(
 157                                         &channel->metadata_cache->contents));
 158
 159         ret = 0;
 160         goto end;
 161
 162 end_free_mutex:
 163         pthread_mutex_destroy(&channel->metadata_cache->lock);
 164 end_free_cache:
 165         free(channel->metadata_cache);
 166 end:
 167         return ret;
 168 }
 169
 170 /*
 171  * Destroy and free the metadata cache
 172  */
 173 void consumer_metadata_cache_destroy(struct lttng_consumer_channel *channel)
 174 {
 175         if (!channel || !channel->metadata_cache) {
 176                 return;
 177         }
 178
 179         DBG("Destroying metadata cache");
 180
 181         pthread_mutex_destroy(&channel->metadata_cache->lock);
 182         lttng_dynamic_buffer_reset(&channel->metadata_cache->contents);
 183         free(channel->metadata_cache);
 184 }
 185
 186 /*
 187  * Check if the cache is flushed up to the offset passed in parameter.
 188  *
 189  * Return 0 if everything has been flushed, 1 if there is data not flushed.
 190  */
 191 int consumer_metadata_cache_flushed(struct lttng_consumer_channel *channel,
 192                 uint64_t offset, int timer)
 193 {
 194         int ret = 0;
 195         struct lttng_consumer_stream *metadata_stream;
 196
 197         LTTNG_ASSERT(channel);
 198         LTTNG_ASSERT(channel->metadata_cache);
 199
 200         /*
 201          * If not called from a timer handler, we have to take the
 202          * channel lock to be mutually exclusive with channel teardown.
 203          * Timer handler does not need to take this lock because it is
 204          * already synchronized by timer stop (and, more importantly,
 205          * taking this lock in a timer handler would cause a deadlock).
 206          */
 207         if (!timer) {
 208                 pthread_mutex_lock(&channel->lock);
 209         }
 210         pthread_mutex_lock(&channel->timer_lock);
 211         metadata_stream = channel->metadata_stream;
 212         if (!metadata_stream) {
 213                 /*
 214                  * Having no metadata stream means the channel is being destroyed so there
 215                  * is no cache to flush anymore.
 216                  */
 217                 ret = 0;
 218                 goto end_unlock_channel;
 219         }
 220
 221         pthread_mutex_lock(&metadata_stream->lock);
 222         pthread_mutex_lock(&channel->metadata_cache->lock);
 223
 224         if (metadata_stream->ust_metadata_pushed >= offset) {
 225                 ret = 0;
 226         } else if (channel->metadata_stream->endpoint_status !=
 227                         CONSUMER_ENDPOINT_ACTIVE) {
 228                 /* An inactive endpoint means we don't have to flush anymore. */
 229                 ret = 0;
 230         } else {
 231                 /* Still not completely flushed. */
 232                 ret = 1;
 233         }
 234
 235         pthread_mutex_unlock(&channel->metadata_cache->lock);
 236         pthread_mutex_unlock(&metadata_stream->lock);
 237 end_unlock_channel:
 238         pthread_mutex_unlock(&channel->timer_lock);
 239         if (!timer) {
 240                 pthread_mutex_unlock(&channel->lock);
 241         }
 242
 243         return ret;
 244 }