Tweak caching topics to fix CEL tests
[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         struct stasis_caching_topic *caching_topic = obj;
62         ast_assert(!stasis_subscription_is_subscribed(caching_topic->sub));
63         ast_assert(stasis_subscription_is_done(caching_topic->sub));
64         ao2_cleanup(caching_topic->sub);
65         caching_topic->sub = NULL;
66         ao2_cleanup(caching_topic->cache);
67         caching_topic->cache = NULL;
68         ao2_cleanup(caching_topic->topic);
69         caching_topic->topic = NULL;
70         ao2_cleanup(caching_topic->original_topic);
71         caching_topic->original_topic = NULL;
72 }
73
74 struct stasis_topic *stasis_caching_get_topic(struct stasis_caching_topic *caching_topic)
75 {
76         return caching_topic->topic;
77 }
78
79 struct stasis_caching_topic *stasis_caching_unsubscribe(struct stasis_caching_topic *caching_topic)
80 {
81         if (caching_topic) {
82                 RAII_VAR(struct stasis_caching_topic *, hold_ref, NULL,
83                         ao2_cleanup);
84
85                 /* The subscription may hold the last reference to this caching
86                  * topic, but we want to make sure the unsubscribe finishes
87                  * before kicking of the caching topic's dtor.
88                  */
89                 ao2_ref(caching_topic, +1);
90                 hold_ref = caching_topic;
91
92                 if (stasis_subscription_is_subscribed(caching_topic->sub)) {
93                         /* Increment the reference to hold on to it past the
94                          * unsubscribe. Will be cleaned up in dtor. */
95                         ao2_ref(caching_topic->sub, +1);
96                         stasis_unsubscribe(caching_topic->sub);
97                 } else {
98                         ast_log(LOG_ERROR, "stasis_caching_topic unsubscribed multiple times\n");
99                 }
100         }
101         return NULL;
102 }
103
104 struct stasis_caching_topic *stasis_caching_unsubscribe_and_join(struct stasis_caching_topic *caching_topic)
105 {
106         if (!caching_topic) {
107                 return NULL;
108         }
109
110         /* Hold a ref past the unsubscribe */
111         ao2_ref(caching_topic, +1);
112         stasis_caching_unsubscribe(caching_topic);
113         stasis_subscription_join(caching_topic->sub);
114         ao2_cleanup(caching_topic);
115         return NULL;
116 }
117
118 struct cache_entry {
119         struct stasis_message_type *type;
120         char *id;
121         struct stasis_message *snapshot;
122 };
123
124 static void cache_entry_dtor(void *obj)
125 {
126         struct cache_entry *entry = obj;
127         ao2_cleanup(entry->type);
128         entry->type = NULL;
129         ast_free(entry->id);
130         entry->id = NULL;
131         ao2_cleanup(entry->snapshot);
132         entry->snapshot = NULL;
133 }
134
135 static struct cache_entry *cache_entry_create(struct stasis_message_type *type, const char *id, struct stasis_message *snapshot)
136 {
137         RAII_VAR(struct cache_entry *, entry, NULL, ao2_cleanup);
138
139         ast_assert(type != NULL);
140         ast_assert(id != NULL);
141
142         entry = ao2_alloc_options(sizeof(*entry), cache_entry_dtor,
143                 AO2_ALLOC_OPT_LOCK_NOLOCK);
144         if (!entry) {
145                 return NULL;
146         }
147
148         entry->id = ast_strdup(id);
149         if (!entry->id) {
150                 return NULL;
151         }
152
153         ao2_ref(type, +1);
154         entry->type = type;
155         if (snapshot != NULL) {
156                 ao2_ref(snapshot, +1);
157                 entry->snapshot = snapshot;
158         }
159
160         ao2_ref(entry, +1);
161         return entry;
162 }
163
164 static int cache_entry_hash(const void *obj, int flags)
165 {
166         const struct cache_entry *entry = obj;
167         int hash = 0;
168
169         ast_assert(!(flags & OBJ_KEY));
170
171         hash += ast_hashtab_hash_string(stasis_message_type_name(entry->type));
172         hash += ast_hashtab_hash_string(entry->id);
173         return hash;
174 }
175
176 static int cache_entry_cmp(void *obj, void *arg, int flags)
177 {
178         const struct cache_entry *left = obj;
179         const struct cache_entry *right = arg;
180
181         ast_assert(!(flags & OBJ_KEY));
182
183         if (left->type == right->type && strcmp(left->id, right->id) == 0) {
184                 return CMP_MATCH | CMP_STOP;
185         }
186
187         return 0;
188 }
189
190 static void cache_dtor(void *obj)
191 {
192         struct stasis_cache *cache = obj;
193
194         ao2_cleanup(cache->entries);
195         cache->entries = NULL;
196 }
197
198 struct stasis_cache *stasis_cache_create(snapshot_get_id id_fn)
199 {
200         RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
201
202         cache = ao2_alloc_options(sizeof(*cache), cache_dtor,
203                 AO2_ALLOC_OPT_LOCK_NOLOCK);
204         if (!cache) {
205                 return NULL;
206         }
207
208         cache->entries = ao2_container_alloc(NUM_CACHE_BUCKETS, cache_entry_hash,
209                 cache_entry_cmp);
210         if (!cache->entries) {
211                 return NULL;
212         }
213
214         cache->id_fn = id_fn;
215
216         ao2_ref(cache, +1);
217         return cache;
218 }
219
220 static struct stasis_message *cache_put(struct stasis_cache *cache,
221         struct stasis_message_type *type, const char *id,
222         struct stasis_message *new_snapshot)
223 {
224         RAII_VAR(struct cache_entry *, new_entry, NULL, ao2_cleanup);
225         RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
226         struct stasis_message *old_snapshot = NULL;
227
228         ast_assert(cache->entries != NULL);
229         ast_assert(new_snapshot == NULL ||
230                 type == stasis_message_type(new_snapshot));
231
232         new_entry = cache_entry_create(type, id, new_snapshot);
233
234         if (new_snapshot == NULL) {
235                 /* Remove entry from cache */
236                 cached_entry = ao2_find(cache->entries, new_entry, OBJ_POINTER | OBJ_UNLINK);
237                 if (cached_entry) {
238                         old_snapshot = cached_entry->snapshot;
239                         cached_entry->snapshot = NULL;
240                 }
241         } else {
242                 /* Insert/update cache */
243                 SCOPED_AO2LOCK(lock, cache->entries);
244
245                 cached_entry = ao2_find(cache->entries, new_entry, OBJ_POINTER | OBJ_NOLOCK);
246                 if (cached_entry) {
247                         /* Update cache. Because objects are moving, no need to update refcounts. */
248                         old_snapshot = cached_entry->snapshot;
249                         cached_entry->snapshot = new_entry->snapshot;
250                         new_entry->snapshot = NULL;
251                 } else {
252                         /* Insert into the cache */
253                         ao2_link_flags(cache->entries, new_entry, OBJ_NOLOCK);
254                 }
255
256         }
257
258         return old_snapshot;
259 }
260
261 struct stasis_message *stasis_cache_get(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
262 {
263         RAII_VAR(struct cache_entry *, search_entry, NULL, ao2_cleanup);
264         RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
265
266         ast_assert(cache->entries != NULL);
267
268         search_entry = cache_entry_create(type, id, NULL);
269         if (search_entry == NULL) {
270                 return NULL;
271         }
272
273         cached_entry = ao2_find(cache->entries, search_entry, OBJ_POINTER);
274         if (cached_entry == NULL) {
275                 return NULL;
276         }
277
278         ast_assert(cached_entry->snapshot != NULL);
279         ao2_ref(cached_entry->snapshot, +1);
280         return cached_entry->snapshot;
281 }
282
283 struct cache_dump_data {
284         struct ao2_container *cached;
285         struct stasis_message_type *type;
286 };
287
288 static int cache_dump_cb(void *obj, void *arg, int flags)
289 {
290         struct cache_dump_data *cache_dump = arg;
291         struct cache_entry *entry = obj;
292
293         if (!cache_dump->type || entry->type == cache_dump->type) {
294                 ao2_link(cache_dump->cached, entry->snapshot);
295         }
296
297         return 0;
298 }
299
300 struct ao2_container *stasis_cache_dump(struct stasis_cache *cache, struct stasis_message_type *type)
301 {
302         struct cache_dump_data cache_dump;
303
304         ast_assert(cache->entries != NULL);
305
306         cache_dump.type = type;
307         cache_dump.cached = ao2_container_alloc_options(
308                 AO2_ALLOC_OPT_LOCK_NOLOCK, 1, NULL, NULL);
309         if (!cache_dump.cached) {
310                 return NULL;
311         }
312
313         ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_cb, &cache_dump);
314         return cache_dump.cached;
315 }
316
317 STASIS_MESSAGE_TYPE_DEFN(stasis_cache_clear_type);
318 STASIS_MESSAGE_TYPE_DEFN(stasis_cache_update_type);
319
320 struct stasis_message *stasis_cache_clear_create(struct stasis_message *id_message)
321 {
322         RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
323
324         msg = stasis_message_create(stasis_cache_clear_type(), id_message);
325         if (!msg) {
326                 return NULL;
327         }
328
329         ao2_ref(msg, +1);
330         return msg;
331 }
332
333 static void stasis_cache_update_dtor(void *obj)
334 {
335         struct stasis_cache_update *update = obj;
336         ao2_cleanup(update->topic);
337         update->topic = NULL;
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_topic *topic, struct stasis_message *old_snapshot, struct stasis_message *new_snapshot)
347 {
348         RAII_VAR(struct stasis_cache_update *, update, NULL, ao2_cleanup);
349         RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
350
351         ast_assert(topic != NULL);
352         ast_assert(old_snapshot != NULL || new_snapshot != NULL);
353
354         update = ao2_alloc_options(sizeof(*update), stasis_cache_update_dtor,
355                 AO2_ALLOC_OPT_LOCK_NOLOCK);
356         if (!update) {
357                 return NULL;
358         }
359
360         ao2_ref(topic, +1);
361         update->topic = topic;
362         if (old_snapshot) {
363                 ao2_ref(old_snapshot, +1);
364                 update->old_snapshot = old_snapshot;
365                 if (!new_snapshot) {
366                         ao2_ref(stasis_message_type(old_snapshot), +1);
367                         update->type = stasis_message_type(old_snapshot);
368                 }
369         }
370         if (new_snapshot) {
371                 ao2_ref(new_snapshot, +1);
372                 update->new_snapshot = new_snapshot;
373                 ao2_ref(stasis_message_type(new_snapshot), +1);
374                 update->type = stasis_message_type(new_snapshot);
375         }
376
377         msg = stasis_message_create(stasis_cache_update_type(), update);
378         if (!msg) {
379                 return NULL;
380         }
381
382         ao2_ref(msg, +1);
383         return msg;
384 }
385
386 static void caching_topic_exec(void *data, struct stasis_subscription *sub,
387         struct stasis_topic *topic, struct stasis_message *message)
388 {
389         RAII_VAR(struct stasis_caching_topic *, caching_topic_needs_unref, NULL, ao2_cleanup);
390         struct stasis_caching_topic *caching_topic = data;
391         const char *id = NULL;
392
393         ast_assert(caching_topic != NULL);
394         ast_assert(caching_topic->topic != NULL);
395         ast_assert(caching_topic->cache != NULL);
396         ast_assert(caching_topic->cache->id_fn != NULL);
397
398         if (stasis_subscription_final_message(sub, message)) {
399                 caching_topic_needs_unref = caching_topic;
400         }
401
402         /* Handle cache clear event */
403         if (stasis_cache_clear_type() == stasis_message_type(message)) {
404                 RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
405                 RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
406                 struct stasis_message *clear_msg = stasis_message_data(message);
407                 const char *clear_id = caching_topic->cache->id_fn(clear_msg);
408                 struct stasis_message_type *clear_type = stasis_message_type(clear_msg);
409
410                 ast_assert(clear_type != NULL);
411
412                 if (clear_id) {
413                         old_snapshot = cache_put(caching_topic->cache, clear_type, clear_id, NULL);
414                         if (old_snapshot) {
415                                 update = update_create(topic, old_snapshot, NULL);
416                                 stasis_publish(caching_topic->topic, update);
417                                 return;
418                         }
419
420                         ast_log(LOG_ERROR,
421                                 "Attempting to remove an item from the %s cache that isn't there: %s %s\n",
422                                 stasis_topic_name(caching_topic->topic), stasis_message_type_name(clear_type), clear_id);
423                         return;
424                 }
425         }
426
427         id = caching_topic->cache->id_fn(message);
428         if (id == NULL) {
429                 /* Object isn't cached; discard */
430         } else {
431                 /* Update the cache */
432                 RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
433                 RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
434
435                 old_snapshot = cache_put(caching_topic->cache, stasis_message_type(message), id, message);
436
437                 update = update_create(topic, old_snapshot, message);
438                 if (update == NULL) {
439                         return;
440                 }
441
442                 stasis_publish(caching_topic->topic, update);
443         }
444 }
445
446 struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *original_topic, struct stasis_cache *cache)
447 {
448         RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, ao2_cleanup);
449         struct stasis_subscription *sub;
450         RAII_VAR(char *, new_name, NULL, free);
451         int ret;
452
453         ret = asprintf(&new_name, "%s-cached", stasis_topic_name(original_topic));
454         if (ret < 0) {
455                 return NULL;
456         }
457
458         caching_topic = ao2_alloc_options(sizeof(*caching_topic),
459                 stasis_caching_topic_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
460         if (caching_topic == NULL) {
461                 return NULL;
462         }
463
464         caching_topic->topic = stasis_topic_create(new_name);
465         if (caching_topic->topic == NULL) {
466                 return NULL;
467         }
468
469         ao2_ref(cache, +1);
470         caching_topic->cache = cache;
471
472         sub = internal_stasis_subscribe(original_topic, caching_topic_exec, caching_topic, 0);
473         if (sub == NULL) {
474                 return NULL;
475         }
476
477         ao2_ref(original_topic, +1);
478         caching_topic->original_topic = original_topic;
479
480         /* This is for the reference contained in the subscription above */
481         ao2_ref(caching_topic, +1);
482         caching_topic->sub = sub;
483
484         /* The subscription holds the reference, so no additional ref bump. */
485         return caching_topic;
486 }
487
488 static void stasis_cache_cleanup(void)
489 {
490         STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_clear_type);
491         STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_update_type);
492 }
493
494 int stasis_cache_init(void)
495 {
496         ast_register_cleanup(stasis_cache_cleanup);
497
498         if (STASIS_MESSAGE_TYPE_INIT(stasis_cache_clear_type) != 0) {
499                 return -1;
500         }
501
502         if (STASIS_MESSAGE_TYPE_INIT(stasis_cache_update_type) != 0) {
503                 return -1;
504         }
505
506         return 0;
507 }
508