src/bin/lttng-relayd/index.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 it
   6  * under the terms of the GNU General Public License, version 2 only, as
   7  * 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 with
  15  * this program; if not, write to the Free Software Foundation, Inc., 51
  16  * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  17  */
  18
  19 #define _GNU_SOURCE
  20 #include <assert.h>
  21
  22 #include <common/common.h>
  23 #include <common/utils.h>
  24
  25 #include "lttng-relayd.h"
  26 #include "index.h"
  27
  28 /*
  29  * Deferred free of a relay index object. MUST only be called by a call RCU.
  30  */
  31 static void deferred_free_relay_index(struct rcu_head *head)
  32 {
  33         struct relay_index *index =
  34                 caa_container_of(head, struct relay_index, rcu_node);
  35
  36         if (index->to_close_fd >= 0) {
  37                 int ret;
  38
  39                 ret = close(index->to_close_fd);
  40                 if (ret < 0) {
  41                         PERROR("Relay index to close fd %d", index->to_close_fd);
  42                 }
  43         }
  44
  45         relay_index_free(index);
  46 }
  47
  48 /*
  49  * Allocate a new relay index object using the given stream ID and sequence
  50  * number as the hash table key.
  51  *
  52  * Return allocated object or else NULL on error.
  53  */
  54 struct relay_index *relay_index_create(uint64_t stream_id,
  55                 uint64_t net_seq_num)
  56 {
  57         struct relay_index *index;
  58
  59         DBG2("Creating relay index with stream id %" PRIu64 " and seqnum %" PRIu64,
  60                         stream_id, net_seq_num);
  61
  62         index = zmalloc(sizeof(*index));
  63         if (index == NULL) {
  64                 PERROR("Relay index zmalloc");
  65                 goto error;
  66         }
  67
  68         index->to_close_fd = -1;
  69         lttng_ht_node_init_two_u64(&index->index_n, stream_id, net_seq_num);
  70
  71 error:
  72         return index;
  73 }
  74
  75 /*
  76  * Find a relayd index in the given hash table.
  77  *
  78  * Return index object or else NULL on error.
  79  */
  80 struct relay_index *relay_index_find(uint64_t stream_id, uint64_t net_seq_num)
  81 {
  82         struct lttng_ht_node_two_u64 *node;
  83         struct lttng_ht_iter iter;
  84         struct lttng_ht_two_u64 key;
  85         struct relay_index *index = NULL;
  86
  87         DBG3("Finding index for stream id %" PRIu64 " and seq_num %" PRIu64,
  88                         stream_id, net_seq_num);
  89
  90         key.key1 = stream_id;
  91         key.key2 = net_seq_num;
  92
  93         lttng_ht_lookup(indexes_ht, (void *)(&key), &iter);
  94         node = lttng_ht_iter_get_node_two_u64(&iter);
  95         if (node == NULL) {
  96                 goto end;
  97         }
  98         index = caa_container_of(node, struct relay_index, index_n);
  99
 100 end:
 101         DBG2("Index %sfound in HT for stream ID %" PRIu64 " and seqnum %" PRIu64,
 102                         (index == NULL) ? "NOT " : "", stream_id, net_seq_num);
 103         return index;
 104 }
 105
 106 /*
 107  * Add unique relay index to the given hash table. In case of a collision, the
 108  * already existing object is put in the given _index variable.
 109  *
 110  * RCU read side lock MUST be acquired.
 111  */
 112 void relay_index_add(struct relay_index *index, struct relay_index **_index)
 113 {
 114         struct cds_lfht_node *node_ptr;
 115
 116         assert(index);
 117
 118         DBG2("Adding relay index with stream id %" PRIu64 " and seqnum %" PRIu64,
 119                         index->key.key1, index->key.key2);
 120
 121         node_ptr = cds_lfht_add_unique(indexes_ht->ht,
 122                         indexes_ht->hash_fct((void *) &index->index_n.key, lttng_ht_seed),
 123                         indexes_ht->match_fct, (void *) &index->index_n.key,
 124                         &index->index_n.node);
 125         if (node_ptr != &index->index_n.node) {
 126                 *_index = caa_container_of(node_ptr, struct relay_index, index_n.node);
 127         }
 128 }
 129
 130 /*
 131  * Write index on disk to the given fd. Once done error or not, it is removed
 132  * from the hash table and destroy the object.
 133  *
 134  * MUST be called with a RCU read side lock held.
 135  *
 136  * Return 0 on success else a negative value.
 137  */
 138 int relay_index_write(int fd, struct relay_index *index)
 139 {
 140         int ret;
 141         struct lttng_ht_iter iter;
 142
 143         DBG2("Writing index for stream ID %" PRIu64 " and seq num %" PRIu64
 144                         " on fd %d", index->key.key1, index->key.key2, fd);
 145
 146         /* Delete index from hash table. */
 147         iter.iter.node = &index->index_n.node;
 148         ret = lttng_ht_del(indexes_ht, &iter);
 149         assert(!ret);
 150         call_rcu(&index->rcu_node, deferred_free_relay_index);
 151
 152         return index_write(fd, &index->index_data, sizeof(index->index_data));
 153 }
 154
 155 /*
 156  * Free the given index.
 157  */
 158 void relay_index_free(struct relay_index *index)
 159 {
 160         free(index);
 161 }
 162
 163 /*
 164  * Safely free the given index using a call RCU.
 165  */
 166 void relay_index_free_safe(struct relay_index *index)
 167 {
 168         if (!index) {
 169                 return;
 170         }
 171
 172         call_rcu(&index->rcu_node, deferred_free_relay_index);
 173 }
 174
 175 /*
 176  * Delete index from the given hash table.
 177  *
 178  * RCU read side lock MUST be acquired.
 179  */
 180 void relay_index_delete(struct relay_index *index)
 181 {
 182         int ret;
 183         struct lttng_ht_iter iter;
 184
 185         DBG3("Relay index with stream ID %" PRIu64 " and seq num %" PRIu64
 186                         "deleted.", index->key.key1, index->key.key2);
 187
 188         /* Delete index from hash table. */
 189         iter.iter.node = &index->index_n.node;
 190         ret = lttng_ht_del(indexes_ht, &iter);
 191         assert(!ret);
 192 }
 193
 194 /*
 195  * Destroy every relay index with the given stream id as part of the key.
 196  */
 197 void relay_index_destroy_by_stream_id(uint64_t stream_id)
 198 {
 199         struct lttng_ht_iter iter;
 200         struct relay_index *index;
 201
 202         rcu_read_lock();
 203         cds_lfht_for_each_entry(indexes_ht->ht, &iter.iter, index, index_n.node) {
 204                 if (index->key.key1 == stream_id) {
 205                         relay_index_delete(index);
 206                         relay_index_free_safe(index);
 207                 }
 208         }
 209         rcu_read_unlock();
 210 }