src/bin/lttng-relayd/index.cpp

   1 /*
   2  * Copyright (C) 2013 Julien Desfossez <jdesfossez@efficios.com>
   3  * Copyright (C) 2013 David Goulet <dgoulet@efficios.com>
   4  * Copyright (C) 2015 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   5  *
   6  * SPDX-License-Identifier: GPL-2.0-only
   7  *
   8  */
   9
  10 #define _LGPL_SOURCE
  11
  12 #include <common/common.hpp>
  13 #include <common/utils.hpp>
  14 #include <common/compat/endian.hpp>
  15
  16 #include "lttng-relayd.hpp"
  17 #include "stream.hpp"
  18 #include "index.hpp"
  19 #include "connection.hpp"
  20
  21 /*
  22  * Allocate a new relay index object. Pass the stream in which it is
  23  * contained as parameter. The sequence number will be used as the hash
  24  * table key.
  25  *
  26  * Called with stream mutex held.
  27  * Return allocated object or else NULL on error.
  28  */
  29 static struct relay_index *relay_index_create(struct relay_stream *stream,
  30                 uint64_t net_seq_num)
  31 {
  32         struct relay_index *index;
  33
  34         DBG2("Creating relay index for stream id %" PRIu64 " and seqnum %" PRIu64,
  35                         stream->stream_handle, net_seq_num);
  36
  37         index = zmalloc<relay_index>();
  38         if (!index) {
  39                 PERROR("Relay index zmalloc");
  40                 goto end;
  41         }
  42         if (!stream_get(stream)) {
  43                 ERR("Cannot get stream");
  44                 free(index);
  45                 index = NULL;
  46                 goto end;
  47         }
  48         index->stream = stream;
  49
  50         lttng_ht_node_init_u64(&index->index_n, net_seq_num);
  51         pthread_mutex_init(&index->lock, NULL);
  52         urcu_ref_init(&index->ref);
  53
  54 end:
  55         return index;
  56 }
  57
  58 /*
  59  * Add unique relay index to the given hash table. In case of a collision, the
  60  * already existing object is put in the given _index variable.
  61  *
  62  * RCU read side lock MUST be acquired.
  63  */
  64 static struct relay_index *relay_index_add_unique(struct relay_stream *stream,
  65                 struct relay_index *index)
  66 {
  67         struct cds_lfht_node *node_ptr;
  68         struct relay_index *_index;
  69
  70         ASSERT_RCU_READ_LOCKED();
  71
  72         DBG2("Adding relay index with stream id %" PRIu64 " and seqnum %" PRIu64,
  73                         stream->stream_handle, index->index_n.key);
  74
  75         node_ptr = cds_lfht_add_unique(stream->indexes_ht->ht,
  76                         stream->indexes_ht->hash_fct(&index->index_n, lttng_ht_seed),
  77                         stream->indexes_ht->match_fct, &index->index_n,
  78                         &index->index_n.node);
  79         if (node_ptr != &index->index_n.node) {
  80                 _index = caa_container_of(node_ptr, struct relay_index,
  81                                 index_n.node);
  82         } else {
  83                 _index = NULL;
  84         }
  85         return _index;
  86 }
  87
  88 /*
  89  * Should be called with RCU read-side lock held.
  90  */
  91 static bool relay_index_get(struct relay_index *index)
  92 {
  93         ASSERT_RCU_READ_LOCKED();
  94
  95         DBG2("index get for stream id %" PRIu64 " and seqnum %" PRIu64 " refcount %d",
  96                         index->stream->stream_handle, index->index_n.key,
  97                         (int) index->ref.refcount);
  98
  99         return urcu_ref_get_unless_zero(&index->ref);
 100 }
 101
 102 /*
 103  * Get a relayd index in within the given stream, or create it if not
 104  * present.
 105  *
 106  * Called with stream mutex held.
 107  * Return index object or else NULL on error.
 108  */
 109 struct relay_index *relay_index_get_by_id_or_create(struct relay_stream *stream,
 110                  uint64_t net_seq_num)
 111 {
 112         struct lttng_ht_node_u64 *node;
 113         struct lttng_ht_iter iter;
 114         struct relay_index *index = NULL;
 115
 116         DBG3("Finding index for stream id %" PRIu64 " and seq_num %" PRIu64,
 117                         stream->stream_handle, net_seq_num);
 118
 119         rcu_read_lock();
 120         lttng_ht_lookup(stream->indexes_ht, &net_seq_num, &iter);
 121         node = lttng_ht_iter_get_node_u64(&iter);
 122         if (node) {
 123                 index = caa_container_of(node, struct relay_index, index_n);
 124         } else {
 125                 struct relay_index *oldindex;
 126
 127                 index = relay_index_create(stream, net_seq_num);
 128                 if (!index) {
 129                         ERR("Cannot create index for stream id %" PRIu64 " and seq_num %" PRIu64,
 130                                 stream->stream_handle, net_seq_num);
 131                         goto end;
 132                 }
 133                 oldindex = relay_index_add_unique(stream, index);
 134                 if (oldindex) {
 135                         /* Added concurrently, keep old. */
 136                         relay_index_put(index);
 137                         index = oldindex;
 138                         if (!relay_index_get(index)) {
 139                                 index = NULL;
 140                         }
 141                 } else {
 142                         stream->indexes_in_flight++;
 143                         index->in_hash_table = true;
 144                 }
 145         }
 146 end:
 147         rcu_read_unlock();
 148         DBG2("Index %sfound or created in HT for stream ID %" PRIu64 " and seqnum %" PRIu64,
 149                         (index == NULL) ? "NOT " : "", stream->stream_handle, net_seq_num);
 150         return index;
 151 }
 152
 153 int relay_index_set_file(struct relay_index *index,
 154                 struct lttng_index_file *index_file,
 155                 uint64_t data_offset)
 156 {
 157         int ret = 0;
 158
 159         pthread_mutex_lock(&index->lock);
 160         if (index->index_file) {
 161                 ret = -1;
 162                 goto end;
 163         }
 164         lttng_index_file_get(index_file);
 165         index->index_file = index_file;
 166         index->index_data.offset = data_offset;
 167 end:
 168         pthread_mutex_unlock(&index->lock);
 169         return ret;
 170 }
 171
 172 int relay_index_set_data(struct relay_index *index,
 173                 const struct ctf_packet_index *data)
 174 {
 175         int ret = 0;
 176
 177         pthread_mutex_lock(&index->lock);
 178         if (index->has_index_data) {
 179                 ret = -1;
 180                 goto end;
 181         }
 182         /* Set everything except data_offset. */
 183         index->index_data.packet_size = data->packet_size;
 184         index->index_data.content_size = data->content_size;
 185         index->index_data.timestamp_begin = data->timestamp_begin;
 186         index->index_data.timestamp_end = data->timestamp_end;
 187         index->index_data.events_discarded = data->events_discarded;
 188         index->index_data.stream_id = data->stream_id;
 189         index->has_index_data = true;
 190 end:
 191         pthread_mutex_unlock(&index->lock);
 192         return ret;
 193 }
 194
 195 static void index_destroy(struct relay_index *index)
 196 {
 197         free(index);
 198 }
 199
 200 static void index_destroy_rcu(struct rcu_head *rcu_head)
 201 {
 202         struct relay_index *index =
 203                 caa_container_of(rcu_head, struct relay_index, rcu_node);
 204
 205         index_destroy(index);
 206 }
 207
 208 /* Stream lock must be held by the caller. */
 209 static void index_release(struct urcu_ref *ref)
 210 {
 211         struct relay_index *index = caa_container_of(ref, struct relay_index, ref);
 212         struct relay_stream *stream = index->stream;
 213         int ret;
 214         struct lttng_ht_iter iter;
 215
 216         if (index->index_file) {
 217                 lttng_index_file_put(index->index_file);
 218                 index->index_file = NULL;
 219         }
 220         if (index->in_hash_table) {
 221                 /* Delete index from hash table. */
 222                 iter.iter.node = &index->index_n.node;
 223                 ret = lttng_ht_del(stream->indexes_ht, &iter);
 224                 LTTNG_ASSERT(!ret);
 225                 stream->indexes_in_flight--;
 226         }
 227
 228         stream_put(index->stream);
 229         index->stream = NULL;
 230
 231         call_rcu(&index->rcu_node, index_destroy_rcu);
 232 }
 233
 234 /*
 235  * Called with stream mutex held.
 236  *
 237  * Stream lock must be held by the caller.
 238  */
 239 void relay_index_put(struct relay_index *index)
 240 {
 241         DBG2("index put for stream id %" PRIu64 " and seqnum %" PRIu64 " refcount %d",
 242                         index->stream->stream_handle, index->index_n.key,
 243                         (int) index->ref.refcount);
 244         /*
 245          * Ensure existence of index->lock for index unlock.
 246          */
 247         rcu_read_lock();
 248         /*
 249          * Index lock ensures that concurrent test and update of stream
 250          * ref is atomic.
 251          */
 252         LTTNG_ASSERT(index->ref.refcount != 0);
 253         urcu_ref_put(&index->ref, index_release);
 254         rcu_read_unlock();
 255 }
 256
 257 /*
 258  * Try to flush index to disk. Releases self-reference to index once
 259  * flush succeeds.
 260  *
 261  * Stream lock must be held by the caller.
 262  * Return 0 on successful flush, a negative value on error, or positive
 263  * value if no flush was performed.
 264  */
 265 int relay_index_try_flush(struct relay_index *index)
 266 {
 267         int ret = 1;
 268         bool flushed = false;
 269
 270         pthread_mutex_lock(&index->lock);
 271         if (index->flushed) {
 272                 goto skip;
 273         }
 274         /* Check if we are ready to flush. */
 275         if (!index->has_index_data || !index->index_file) {
 276                 goto skip;
 277         }
 278
 279         DBG2("Writing index for stream ID %" PRIu64 " and seq num %" PRIu64,
 280                         index->stream->stream_handle, index->index_n.key);
 281         flushed = true;
 282         index->flushed = true;
 283         ret = lttng_index_file_write(index->index_file, &index->index_data);
 284 skip:
 285         pthread_mutex_unlock(&index->lock);
 286
 287         if (flushed) {
 288                 /* Put self-ref from index now that it has been flushed. */
 289                 relay_index_put(index);
 290         }
 291         return ret;
 292 }
 293
 294 /*
 295  * Close every relay index within a given stream, without flushing
 296  * them.
 297  */
 298 void relay_index_close_all(struct relay_stream *stream)
 299 {
 300         struct lttng_ht_iter iter;
 301         struct relay_index *index;
 302
 303         rcu_read_lock();
 304         cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter,
 305                         index, index_n.node) {
 306                 /* Put self-ref from index. */
 307                 relay_index_put(index);
 308         }
 309         rcu_read_unlock();
 310 }
 311
 312 void relay_index_close_partial_fd(struct relay_stream *stream)
 313 {
 314         struct lttng_ht_iter iter;
 315         struct relay_index *index;
 316
 317         rcu_read_lock();
 318         cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter,
 319                         index, index_n.node) {
 320                 if (!index->index_file) {
 321                         continue;
 322                 }
 323                 /*
 324                  * Partial index has its index_file: we have only
 325                  * received its info from the data socket.
 326                  * Put self-ref from index.
 327                  */
 328                 relay_index_put(index);
 329         }
 330         rcu_read_unlock();
 331 }
 332
 333 uint64_t relay_index_find_last(struct relay_stream *stream)
 334 {
 335         struct lttng_ht_iter iter;
 336         struct relay_index *index;
 337         uint64_t net_seq_num = -1ULL;
 338
 339         rcu_read_lock();
 340         cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter,
 341                         index, index_n.node) {
 342                 if (net_seq_num == -1ULL ||
 343                                 index->index_n.key > net_seq_num) {
 344                         net_seq_num = index->index_n.key;
 345                 }
 346         }
 347         rcu_read_unlock();
 348         return net_seq_num;
 349 }
 350
 351 /*
 352  * Update the index file of an already existing relay_index.
 353  * Offsets by 'removed_data_count' the offset field of an index.
 354  */
 355 static
 356 int relay_index_switch_file(struct relay_index *index,
 357                 struct lttng_index_file *new_index_file,
 358                 uint64_t removed_data_count)
 359 {
 360         int ret = 0;
 361         uint64_t offset;
 362
 363         pthread_mutex_lock(&index->lock);
 364         if (!index->index_file) {
 365                 ERR("No index_file");
 366                 ret = 0;
 367                 goto end;
 368         }
 369
 370         lttng_index_file_put(index->index_file);
 371         lttng_index_file_get(new_index_file);
 372         index->index_file = new_index_file;
 373         offset = be64toh(index->index_data.offset);
 374         index->index_data.offset = htobe64(offset - removed_data_count);
 375
 376 end:
 377         pthread_mutex_unlock(&index->lock);
 378         return ret;
 379 }
 380
 381 /*
 382  * Switch the index file of all pending indexes for a stream and update the
 383  * data offset by substracting the last safe position.
 384  * Stream lock must be held.
 385  */
 386 int relay_index_switch_all_files(struct relay_stream *stream)
 387 {
 388         struct lttng_ht_iter iter;
 389         struct relay_index *index;
 390         int ret = 0;
 391
 392         rcu_read_lock();
 393         cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter,
 394                         index, index_n.node) {
 395                 ret = relay_index_switch_file(index, stream->index_file,
 396                                 stream->pos_after_last_complete_data_index);
 397                 if (ret) {
 398                         goto end;
 399                 }
 400         }
 401 end:
 402         rcu_read_unlock();
 403         return ret;
 404 }
 405
 406 /*
 407  * Set index data from the control port to a given index object.
 408  */
 409 int relay_index_set_control_data(struct relay_index *index,
 410                 const struct lttcomm_relayd_index *data,
 411                 unsigned int minor_version)
 412 {
 413         /* The index on disk is encoded in big endian. */
 414         ctf_packet_index index_data {};
 415
 416         index_data.packet_size = htobe64(data->packet_size);
 417         index_data.content_size = htobe64(data->content_size);
 418         index_data.timestamp_begin = htobe64(data->timestamp_begin);
 419         index_data.timestamp_end = htobe64(data->timestamp_end);
 420         index_data.events_discarded = htobe64(data->events_discarded);
 421         index_data.stream_id = htobe64(data->stream_id);
 422
 423         if (minor_version >= 8) {
 424                 index->index_data.stream_instance_id = htobe64(data->stream_instance_id);
 425                 index->index_data.packet_seq_num = htobe64(data->packet_seq_num);
 426         } else {
 427                 uint64_t unset_value = -1ULL;
 428
 429                 index->index_data.stream_instance_id = htobe64(unset_value);
 430                 index->index_data.packet_seq_num = htobe64(unset_value);
 431         }
 432
 433         return relay_index_set_data(index, &index_data);
 434 }