src/common/consumer-metadata-cache.c

   1 /*
   2  * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
   3  *                      David Goulet <dgoulet@efficios.com>
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License, version 2 only,
   7  * as published by the Free Software Foundation.
   8  *
   9  * This program is distributed in the hope that it will be useful, but WITHOUT
  10  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  11  * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
  12  * more details.
  13  *
  14  * You should have received a copy of the GNU General Public License along
  15  * with this program; if not, write to the Free Software Foundation, Inc.,
  16  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  17  */
  18
  19 #define _GNU_SOURCE
  20 #include <assert.h>
  21 #include <pthread.h>
  22 #include <stdlib.h>
  23 #include <string.h>
  24 #include <sys/types.h>
  25 #include <unistd.h>
  26 #include <inttypes.h>
  27
  28 #include <common/common.h>
  29 #include <common/utils.h>
  30 #include <common/sessiond-comm/sessiond-comm.h>
  31 #include <common/ust-consumer/ust-consumer.h>
  32 #include <common/consumer.h>
  33
  34 #include "consumer-metadata-cache.h"
  35
  36 /*
  37  * Extend the allocated size of the metadata cache. Called only from
  38  * lttng_ustconsumer_write_metadata_cache.
  39  *
  40  * Return 0 on success, a negative value on error.
  41  */
  42 static int extend_metadata_cache(struct lttng_consumer_channel *channel,
  43                 unsigned int size)
  44 {
  45         int ret = 0;
  46         char *tmp_data_ptr;
  47         unsigned int new_size;
  48
  49         assert(channel);
  50         assert(channel->metadata_cache);
  51
  52         new_size = max_t(unsigned int,
  53                         channel->metadata_cache->cache_alloc_size + size,
  54                         channel->metadata_cache->cache_alloc_size << 1);
  55         DBG("Extending metadata cache to %u", new_size);
  56         tmp_data_ptr = realloc(channel->metadata_cache->data, new_size);
  57         if (!tmp_data_ptr) {
  58                 ERR("Reallocating metadata cache");
  59                 free(channel->metadata_cache->data);
  60                 ret = -1;
  61                 goto end;
  62         }
  63         channel->metadata_cache->data = tmp_data_ptr;
  64         channel->metadata_cache->cache_alloc_size = new_size;
  65
  66 end:
  67         return ret;
  68 }
  69
  70 /*
  71  * Write metadata to the cache, extend the cache if necessary. We support
  72  * non-contiguous updates but not overlapping ones. If there is contiguous
  73  * metadata in the cache, we send it to the ring buffer. The metadata cache
  74  * lock MUST be acquired to write in the cache.
  75  *
  76  * Return 0 on success, a negative value on error.
  77  */
  78 int consumer_metadata_cache_write(struct lttng_consumer_channel *channel,
  79                 unsigned int offset, unsigned int len, char *data)
  80 {
  81         int ret = 0;
  82         struct consumer_metadata_cache *cache;
  83
  84         assert(channel);
  85         assert(channel->metadata_cache);
  86
  87         cache = channel->metadata_cache;
  88         DBG("Writing %u bytes from offset %u in metadata cache", len, offset);
  89
  90         if (offset + len > cache->cache_alloc_size) {
  91                 ret = extend_metadata_cache(channel,
  92                                 len - cache->cache_alloc_size + offset);
  93                 if (ret < 0) {
  94                         ERR("Extending metadata cache");
  95                         goto end;
  96                 }
  97         }
  98
  99         memcpy(cache->data + offset, data, len);
 100         cache->total_bytes_written += len;
 101         if (offset + len > cache->max_offset) {
 102                 cache->max_offset = offset + len;
 103         }
 104
 105         if (cache->max_offset == cache->total_bytes_written) {
 106                 offset = cache->rb_pushed;
 107                 len = cache->total_bytes_written - cache->rb_pushed;
 108                 ret = lttng_ustconsumer_push_metadata(channel, cache->data, offset,
 109                                 len);
 110                 if (ret < 0) {
 111                         ERR("Pushing metadata");
 112                         goto end;
 113                 }
 114                 cache->rb_pushed += len;
 115         }
 116
 117 end:
 118         return ret;
 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         assert(channel);
 131
 132         channel->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         channel->metadata_cache->cache_alloc_size = DEFAULT_METADATA_CACHE_SIZE;
 146         channel->metadata_cache->data = zmalloc(
 147                         channel->metadata_cache->cache_alloc_size * sizeof(char));
 148         if (!channel->metadata_cache->data) {
 149                 PERROR("zmalloc metadata cache data");
 150                 ret = -1;
 151                 goto end_free_mutex;
 152         }
 153         DBG("Allocated metadata cache of %" PRIu64 " bytes",
 154                         channel->metadata_cache->cache_alloc_size);
 155
 156         ret = 0;
 157         goto end;
 158
 159 end_free_mutex:
 160         pthread_mutex_destroy(&channel->metadata_cache->lock);
 161 end_free_cache:
 162         free(channel->metadata_cache);
 163 end:
 164         return ret;
 165 }
 166
 167 /*
 168  * Destroy and free the metadata cache
 169  */
 170 void consumer_metadata_cache_destroy(struct lttng_consumer_channel *channel)
 171 {
 172         if (!channel || !channel->metadata_cache) {
 173                 return;
 174         }
 175
 176         DBG("Destroying metadata cache");
 177
 178         if (channel->metadata_cache->max_offset >
 179                         channel->metadata_cache->rb_pushed) {
 180                 ERR("Destroying a cache not entirely commited");
 181         }
 182
 183         pthread_mutex_destroy(&channel->metadata_cache->lock);
 184         free(channel->metadata_cache->data);
 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)
 195 {
 196         int ret;
 197         struct consumer_metadata_cache *cache;
 198
 199         assert(channel);
 200         assert(channel->metadata_cache);
 201
 202         cache = channel->metadata_cache;
 203
 204         pthread_mutex_lock(&channel->metadata_cache->lock);
 205         if (cache->rb_pushed >= offset) {
 206                 ret = 0;
 207         } else {
 208                 ret = 1;
 209         }
 210         pthread_mutex_unlock(&channel->metadata_cache->lock);
 211
 212         return ret;
 213 }