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.h>
  18 #include <common/utils.h>
  19 #include <common/sessiond-comm/sessiond-comm.h>
  20 #include <common/ust-consumer/ust-consumer.h>
  21 #include <common/consumer/consumer.h>
  22
  23 #include "consumer-metadata-cache.h"
  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 = (consumer_metadata_cache *) zmalloc(
 133                         sizeof(struct consumer_metadata_cache));
 134         if (!channel->metadata_cache) {
 135                 PERROR("zmalloc metadata cache struct");
 136                 ret = -1;
 137                 goto end;
 138         }
 139         ret = pthread_mutex_init(&channel->metadata_cache->lock, NULL);
 140         if (ret != 0) {
 141                 PERROR("mutex init");
 142                 goto end_free_cache;
 143         }
 144
 145         lttng_dynamic_buffer_init(&channel->metadata_cache->contents);
 146         ret = lttng_dynamic_buffer_set_capacity(
 147                         &channel->metadata_cache->contents,
 148                         DEFAULT_METADATA_CACHE_SIZE);
 149         if (ret) {
 150                 PERROR("Failed to pre-allocate metadata cache storage of %d bytes on creation",
 151                                 DEFAULT_METADATA_CACHE_SIZE);
 152                 ret = -1;
 153                 goto end_free_mutex;
 154         }
 155
 156         DBG("Allocated metadata cache: current capacity = %zu",
 157                         lttng_dynamic_buffer_get_capacity_left(
 158                                         &channel->metadata_cache->contents));
 159
 160         ret = 0;
 161         goto end;
 162
 163 end_free_mutex:
 164         pthread_mutex_destroy(&channel->metadata_cache->lock);
 165 end_free_cache:
 166         free(channel->metadata_cache);
 167 end:
 168         return ret;
 169 }
 170
 171 /*
 172  * Destroy and free the metadata cache
 173  */
 174 void consumer_metadata_cache_destroy(struct lttng_consumer_channel *channel)
 175 {
 176         if (!channel || !channel->metadata_cache) {
 177                 return;
 178         }
 179
 180         DBG("Destroying metadata cache");
 181
 182         pthread_mutex_destroy(&channel->metadata_cache->lock);
 183         lttng_dynamic_buffer_reset(&channel->metadata_cache->contents);
 184         free(channel->metadata_cache);
 185 }
 186
 187 /*
 188  * Check if the cache is flushed up to the offset passed in parameter.
 189  *
 190  * Return 0 if everything has been flushed, 1 if there is data not flushed.
 191  */
 192 int consumer_metadata_cache_flushed(struct lttng_consumer_channel *channel,
 193                 uint64_t offset, int timer)
 194 {
 195         int ret = 0;
 196         struct lttng_consumer_stream *metadata_stream;
 197
 198         LTTNG_ASSERT(channel);
 199         LTTNG_ASSERT(channel->metadata_cache);
 200
 201         /*
 202          * If not called from a timer handler, we have to take the
 203          * channel lock to be mutually exclusive with channel teardown.
 204          * Timer handler does not need to take this lock because it is
 205          * already synchronized by timer stop (and, more importantly,
 206          * taking this lock in a timer handler would cause a deadlock).
 207          */
 208         if (!timer) {
 209                 pthread_mutex_lock(&channel->lock);
 210         }
 211         pthread_mutex_lock(&channel->timer_lock);
 212         metadata_stream = channel->metadata_stream;
 213         if (!metadata_stream) {
 214                 /*
 215                  * Having no metadata stream means the channel is being destroyed so there
 216                  * is no cache to flush anymore.
 217                  */
 218                 ret = 0;
 219                 goto end_unlock_channel;
 220         }
 221
 222         pthread_mutex_lock(&metadata_stream->lock);
 223         pthread_mutex_lock(&channel->metadata_cache->lock);
 224
 225         if (metadata_stream->ust_metadata_pushed >= offset) {
 226                 ret = 0;
 227         } else if (channel->metadata_stream->endpoint_status !=
 228                         CONSUMER_ENDPOINT_ACTIVE) {
 229                 /* An inactive endpoint means we don't have to flush anymore. */
 230                 ret = 0;
 231         } else {
 232                 /* Still not completely flushed. */
 233                 ret = 1;
 234         }
 235
 236         pthread_mutex_unlock(&channel->metadata_cache->lock);
 237         pthread_mutex_unlock(&metadata_stream->lock);
 238 end_unlock_channel:
 239         pthread_mutex_unlock(&channel->timer_lock);
 240         if (!timer) {
 241                 pthread_mutex_unlock(&channel->lock);
 242         }
 243
 244         return ret;
 245 }