stasis_cache.c: Remove some unnecessary RAII_VAR() usage.
[asterisk/asterisk.git] / main / stasis_cache.c
1 /*
2  * Asterisk -- An open source telephony toolkit.
3  *
4  * Copyright (C) 2013, Digium, Inc.
5  *
6  * David M. Lee, II <dlee@digium.com>
7  *
8  * See http://www.asterisk.org for more information about
9  * the Asterisk project. Please do not directly contact
10  * any of the maintainers of this project for assistance;
11  * the project provides a web site, mailing lists and IRC
12  * channels for your use.
13  *
14  * This program is free software, distributed under the terms of
15  * the GNU General Public License Version 2. See the LICENSE file
16  * at the top of the source tree.
17  */
18
19 /*! \file
20  *
21  * \brief Stasis Message API.
22  *
23  * \author David M. Lee, II <dlee@digium.com>
24  */
25
26 /*** MODULEINFO
27         <support_level>core</support_level>
28  ***/
29
30 #include "asterisk.h"
31
32 ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
33
34 #include "asterisk/astobj2.h"
35 #include "asterisk/hashtab.h"
36 #include "asterisk/stasis_internal.h"
37 #include "asterisk/stasis.h"
38 #include "asterisk/utils.h"
39
40 #ifdef LOW_MEMORY
41 #define NUM_CACHE_BUCKETS 17
42 #else
43 #define NUM_CACHE_BUCKETS 563
44 #endif
45
46 /*! \internal */
47 struct stasis_cache {
48         struct ao2_container *entries;
49         snapshot_get_id id_fn;
50 };
51
52 /*! \internal */
53 struct stasis_caching_topic {
54         struct stasis_cache *cache;
55         struct stasis_topic *topic;
56         struct stasis_topic *original_topic;
57         struct stasis_subscription *sub;
58 };
59
60 static void stasis_caching_topic_dtor(void *obj)
61 {
62         struct stasis_caching_topic *caching_topic = obj;
63
64         /* Caching topics contain subscriptions, and must be manually
65          * unsubscribed. */
66         ast_assert(!stasis_subscription_is_subscribed(caching_topic->sub));
67         /* If there are any messages in flight to this subscription; that would
68          * be bad. */
69         ast_assert(stasis_subscription_is_done(caching_topic->sub));
70
71         ao2_cleanup(caching_topic->sub);
72         caching_topic->sub = NULL;
73         ao2_cleanup(caching_topic->cache);
74         caching_topic->cache = NULL;
75         ao2_cleanup(caching_topic->topic);
76         caching_topic->topic = NULL;
77         ao2_cleanup(caching_topic->original_topic);
78         caching_topic->original_topic = NULL;
79 }
80
81 struct stasis_topic *stasis_caching_get_topic(struct stasis_caching_topic *caching_topic)
82 {
83         return caching_topic->topic;
84 }
85
86 struct stasis_caching_topic *stasis_caching_unsubscribe(struct stasis_caching_topic *caching_topic)
87 {
88         if (!caching_topic) {
89                 return NULL;
90         }
91
92         /*
93          * The subscription may hold the last reference to this caching
94          * topic, but we want to make sure the unsubscribe finishes
95          * before kicking of the caching topic's dtor.
96          */
97         ao2_ref(caching_topic, +1);
98
99         if (stasis_subscription_is_subscribed(caching_topic->sub)) {
100                 /*
101                  * Increment the reference to hold on to it past the
102                  * unsubscribe. Will be cleaned up in dtor.
103                  */
104                 ao2_ref(caching_topic->sub, +1);
105                 stasis_unsubscribe(caching_topic->sub);
106         } else {
107                 ast_log(LOG_ERROR, "stasis_caching_topic unsubscribed multiple times\n");
108         }
109         ao2_cleanup(caching_topic);
110         return NULL;
111 }
112
113 struct stasis_caching_topic *stasis_caching_unsubscribe_and_join(struct stasis_caching_topic *caching_topic)
114 {
115         if (!caching_topic) {
116                 return NULL;
117         }
118
119         /* Hold a ref past the unsubscribe */
120         ao2_ref(caching_topic, +1);
121         stasis_caching_unsubscribe(caching_topic);
122         stasis_subscription_join(caching_topic->sub);
123         ao2_cleanup(caching_topic);
124         return NULL;
125 }
126
127 struct cache_entry {
128         struct stasis_message_type *type;
129         char *id;
130         struct stasis_message *snapshot;
131 };
132
133 static void cache_entry_dtor(void *obj)
134 {
135         struct cache_entry *entry = obj;
136         ao2_cleanup(entry->type);
137         entry->type = NULL;
138         ast_free(entry->id);
139         entry->id = NULL;
140         ao2_cleanup(entry->snapshot);
141         entry->snapshot = NULL;
142 }
143
144 static struct cache_entry *cache_entry_create(struct stasis_message_type *type, const char *id, struct stasis_message *snapshot)
145 {
146         RAII_VAR(struct cache_entry *, entry, NULL, ao2_cleanup);
147
148         ast_assert(type != NULL);
149         ast_assert(id != NULL);
150
151         entry = ao2_alloc_options(sizeof(*entry), cache_entry_dtor,
152                 AO2_ALLOC_OPT_LOCK_NOLOCK);
153         if (!entry) {
154                 return NULL;
155         }
156
157         entry->id = ast_strdup(id);
158         if (!entry->id) {
159                 return NULL;
160         }
161
162         ao2_ref(type, +1);
163         entry->type = type;
164         if (snapshot != NULL) {
165                 ao2_ref(snapshot, +1);
166                 entry->snapshot = snapshot;
167         }
168
169         ao2_ref(entry, +1);
170         return entry;
171 }
172
173 static int cache_entry_hash(const void *obj, int flags)
174 {
175         const struct cache_entry *entry = obj;
176         int hash = 0;
177
178         ast_assert(!(flags & OBJ_KEY));
179
180         hash += ast_hashtab_hash_string(stasis_message_type_name(entry->type));
181         hash += ast_hashtab_hash_string(entry->id);
182         return hash;
183 }
184
185 static int cache_entry_cmp(void *obj, void *arg, int flags)
186 {
187         const struct cache_entry *left = obj;
188         const struct cache_entry *right = arg;
189
190         ast_assert(!(flags & OBJ_KEY));
191
192         if (left->type == right->type && strcmp(left->id, right->id) == 0) {
193                 return CMP_MATCH | CMP_STOP;
194         }
195
196         return 0;
197 }
198
199 static void cache_dtor(void *obj)
200 {
201         struct stasis_cache *cache = obj;
202
203         ao2_cleanup(cache->entries);
204         cache->entries = NULL;
205 }
206
207 struct stasis_cache *stasis_cache_create(snapshot_get_id id_fn)
208 {
209         RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
210
211         cache = ao2_alloc_options(sizeof(*cache), cache_dtor,
212                 AO2_ALLOC_OPT_LOCK_NOLOCK);
213         if (!cache) {
214                 return NULL;
215         }
216
217         cache->entries = ao2_container_alloc(NUM_CACHE_BUCKETS, cache_entry_hash,
218                 cache_entry_cmp);
219         if (!cache->entries) {
220                 return NULL;
221         }
222
223         cache->id_fn = id_fn;
224
225         ao2_ref(cache, +1);
226         return cache;
227 }
228
229 static struct stasis_message *cache_put(struct stasis_cache *cache,
230         struct stasis_message_type *type, const char *id,
231         struct stasis_message *new_snapshot)
232 {
233         RAII_VAR(struct cache_entry *, new_entry, NULL, ao2_cleanup);
234         RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
235         struct stasis_message *old_snapshot = NULL;
236
237         ast_assert(cache->entries != NULL);
238         ast_assert(new_snapshot == NULL ||
239                 type == stasis_message_type(new_snapshot));
240
241         new_entry = cache_entry_create(type, id, new_snapshot);
242
243         if (new_snapshot == NULL) {
244                 /* Remove entry from cache */
245                 cached_entry = ao2_find(cache->entries, new_entry, OBJ_POINTER | OBJ_UNLINK);
246                 if (cached_entry) {
247                         old_snapshot = cached_entry->snapshot;
248                         cached_entry->snapshot = NULL;
249                 }
250         } else {
251                 /* Insert/update cache */
252                 SCOPED_AO2LOCK(lock, cache->entries);
253
254                 cached_entry = ao2_find(cache->entries, new_entry, OBJ_POINTER | OBJ_NOLOCK);
255                 if (cached_entry) {
256                         /* Update cache. Because objects are moving, no need to update refcounts. */
257                         old_snapshot = cached_entry->snapshot;
258                         cached_entry->snapshot = new_entry->snapshot;
259                         new_entry->snapshot = NULL;
260                 } else {
261                         /* Insert into the cache */
262                         ao2_link_flags(cache->entries, new_entry, OBJ_NOLOCK);
263                 }
264
265         }
266
267         return old_snapshot;
268 }
269
270 struct stasis_message *stasis_cache_get(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
271 {
272         RAII_VAR(struct cache_entry *, search_entry, NULL, ao2_cleanup);
273         RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
274
275         ast_assert(cache->entries != NULL);
276
277         search_entry = cache_entry_create(type, id, NULL);
278         if (search_entry == NULL) {
279                 return NULL;
280         }
281
282         cached_entry = ao2_find(cache->entries, search_entry, OBJ_POINTER);
283         if (cached_entry == NULL) {
284                 return NULL;
285         }
286
287         ast_assert(cached_entry->snapshot != NULL);
288         ao2_ref(cached_entry->snapshot, +1);
289         return cached_entry->snapshot;
290 }
291
292 struct cache_dump_data {
293         struct ao2_container *cached;
294         struct stasis_message_type *type;
295 };
296
297 static int cache_dump_cb(void *obj, void *arg, int flags)
298 {
299         struct cache_dump_data *cache_dump = arg;
300         struct cache_entry *entry = obj;
301
302         if (!cache_dump->type || entry->type == cache_dump->type) {
303                 ao2_link(cache_dump->cached, entry->snapshot);
304         }
305
306         return 0;
307 }
308
309 struct ao2_container *stasis_cache_dump(struct stasis_cache *cache, struct stasis_message_type *type)
310 {
311         struct cache_dump_data cache_dump;
312
313         ast_assert(cache->entries != NULL);
314
315         cache_dump.type = type;
316         cache_dump.cached = ao2_container_alloc_options(
317                 AO2_ALLOC_OPT_LOCK_NOLOCK, 1, NULL, NULL);
318         if (!cache_dump.cached) {
319                 return NULL;
320         }
321
322         ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_cb, &cache_dump);
323         return cache_dump.cached;
324 }
325
326 STASIS_MESSAGE_TYPE_DEFN(stasis_cache_clear_type);
327 STASIS_MESSAGE_TYPE_DEFN(stasis_cache_update_type);
328
329 struct stasis_message *stasis_cache_clear_create(struct stasis_message *id_message)
330 {
331         return stasis_message_create(stasis_cache_clear_type(), id_message);
332 }
333
334 static void stasis_cache_update_dtor(void *obj)
335 {
336         struct stasis_cache_update *update = obj;
337
338         ao2_cleanup(update->old_snapshot);
339         update->old_snapshot = NULL;
340         ao2_cleanup(update->new_snapshot);
341         update->new_snapshot = NULL;
342         ao2_cleanup(update->type);
343         update->type = NULL;
344 }
345
346 static struct stasis_message *update_create(struct stasis_message *old_snapshot, struct stasis_message *new_snapshot)
347 {
348         struct stasis_cache_update *update;
349         struct stasis_message *msg;
350
351         ast_assert(old_snapshot != NULL || new_snapshot != NULL);
352
353         update = ao2_alloc_options(sizeof(*update), stasis_cache_update_dtor,
354                 AO2_ALLOC_OPT_LOCK_NOLOCK);
355         if (!update) {
356                 return NULL;
357         }
358
359         if (old_snapshot) {
360                 ao2_ref(old_snapshot, +1);
361                 update->old_snapshot = old_snapshot;
362                 if (!new_snapshot) {
363                         ao2_ref(stasis_message_type(old_snapshot), +1);
364                         update->type = stasis_message_type(old_snapshot);
365                 }
366         }
367         if (new_snapshot) {
368                 ao2_ref(new_snapshot, +1);
369                 update->new_snapshot = new_snapshot;
370                 ao2_ref(stasis_message_type(new_snapshot), +1);
371                 update->type = stasis_message_type(new_snapshot);
372         }
373
374         msg = stasis_message_create(stasis_cache_update_type(), update);
375
376         ao2_cleanup(update);
377         return msg;
378 }
379
380 static void caching_topic_exec(void *data, struct stasis_subscription *sub,
381         struct stasis_message *message)
382 {
383         RAII_VAR(struct stasis_caching_topic *, caching_topic_needs_unref, NULL, ao2_cleanup);
384         struct stasis_caching_topic *caching_topic = data;
385         const char *id = NULL;
386
387         ast_assert(caching_topic != NULL);
388         ast_assert(caching_topic->topic != NULL);
389         ast_assert(caching_topic->cache != NULL);
390         ast_assert(caching_topic->cache->id_fn != NULL);
391
392         if (stasis_subscription_final_message(sub, message)) {
393                 caching_topic_needs_unref = caching_topic;
394         }
395
396         /* Handle cache clear event */
397         if (stasis_cache_clear_type() == stasis_message_type(message)) {
398                 RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
399                 RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
400                 struct stasis_message *clear_msg = stasis_message_data(message);
401                 const char *clear_id = caching_topic->cache->id_fn(clear_msg);
402                 struct stasis_message_type *clear_type = stasis_message_type(clear_msg);
403
404                 ast_assert(clear_type != NULL);
405
406                 if (clear_id) {
407                         old_snapshot = cache_put(caching_topic->cache, clear_type, clear_id, NULL);
408                         if (old_snapshot) {
409                                 update = update_create(old_snapshot, NULL);
410                                 stasis_publish(caching_topic->topic, update);
411                                 return;
412                         }
413
414                         ast_log(LOG_ERROR,
415                                 "Attempting to remove an item from the %s cache that isn't there: %s %s\n",
416                                 stasis_topic_name(caching_topic->topic), stasis_message_type_name(clear_type), clear_id);
417                         return;
418                 }
419         }
420
421         id = caching_topic->cache->id_fn(message);
422         if (id == NULL) {
423                 /* Object isn't cached; discard */
424         } else {
425                 /* Update the cache */
426                 RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
427                 RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
428
429                 old_snapshot = cache_put(caching_topic->cache, stasis_message_type(message), id, message);
430
431                 update = update_create(old_snapshot, message);
432                 if (update == NULL) {
433                         return;
434                 }
435
436                 stasis_publish(caching_topic->topic, update);
437         }
438 }
439
440 struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *original_topic, struct stasis_cache *cache)
441 {
442         RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, ao2_cleanup);
443         struct stasis_subscription *sub;
444         RAII_VAR(char *, new_name, NULL, ast_free);
445         int ret;
446
447         ret = ast_asprintf(&new_name, "%s-cached", stasis_topic_name(original_topic));
448         if (ret < 0) {
449                 return NULL;
450         }
451
452         caching_topic = ao2_alloc_options(sizeof(*caching_topic),
453                 stasis_caching_topic_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
454         if (caching_topic == NULL) {
455                 return NULL;
456         }
457
458         caching_topic->topic = stasis_topic_create(new_name);
459         if (caching_topic->topic == NULL) {
460                 return NULL;
461         }
462
463         ao2_ref(cache, +1);
464         caching_topic->cache = cache;
465
466         sub = internal_stasis_subscribe(original_topic, caching_topic_exec, caching_topic, 0);
467         if (sub == NULL) {
468                 return NULL;
469         }
470
471         ao2_ref(original_topic, +1);
472         caching_topic->original_topic = original_topic;
473
474         /* This is for the reference contained in the subscription above */
475         ao2_ref(caching_topic, +1);
476         caching_topic->sub = sub;
477
478         /* The subscription holds the reference, so no additional ref bump. */
479         return caching_topic;
480 }
481
482 static void stasis_cache_cleanup(void)
483 {
484         STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_clear_type);
485         STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_update_type);
486 }
487
488 int stasis_cache_init(void)
489 {
490         ast_register_cleanup(stasis_cache_cleanup);
491
492         if (STASIS_MESSAGE_TYPE_INIT(stasis_cache_clear_type) != 0) {
493                 return -1;
494         }
495
496         if (STASIS_MESSAGE_TYPE_INIT(stasis_cache_update_type) != 0) {
497                 return -1;
498         }
499
500         return 0;
501 }
502