8cacdd2f01eb8807d9a5d5de239277b311776256
[lttng-tools.git] / 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 }
This page took 0.034104 seconds and 4 git commands to generate.