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

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