a82492870446dd5230ba7746556a0cc6ad2a75a3
[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.h"
37 #include "asterisk/utils.h"
38
39 #ifdef LOW_MEMORY
40 #define NUM_CACHE_BUCKETS 17
41 #else
42 #define NUM_CACHE_BUCKETS 563
43 #endif
44
45 /*! \private */
46 struct stasis_caching_topic {
47         struct ao2_container *cache;
48         struct stasis_topic *topic;
49         struct stasis_subscription *sub;
50         snapshot_get_id id_fn;
51 };
52
53 static void stasis_caching_topic_dtor(void *obj) {
54         struct stasis_caching_topic *caching_topic = obj;
55         ast_assert(!stasis_subscription_is_subscribed(caching_topic->sub));
56         caching_topic->sub = NULL;
57         ao2_cleanup(caching_topic->cache);
58         caching_topic->cache = NULL;
59         ao2_cleanup(caching_topic->topic);
60         caching_topic->topic = NULL;
61 }
62
63 struct stasis_topic *stasis_caching_get_topic(struct stasis_caching_topic *caching_topic)
64 {
65         return caching_topic->topic;
66 }
67
68 struct stasis_caching_topic *stasis_caching_unsubscribe(struct stasis_caching_topic *caching_topic)
69 {
70         if (caching_topic) {
71                 if (stasis_subscription_is_subscribed(caching_topic->sub)) {
72                         stasis_unsubscribe(caching_topic->sub);
73                 } else {
74                         ast_log(LOG_ERROR, "stasis_caching_topic unsubscribed multiple times\n");
75                 }
76         }
77         return NULL;
78 }
79
80 struct cache_entry {
81         struct stasis_message_type *type;
82         char *id;
83         struct stasis_message *snapshot;
84 };
85
86 static void cache_entry_dtor(void *obj)
87 {
88         struct cache_entry *entry = obj;
89         ao2_cleanup(entry->type);
90         entry->type = NULL;
91         ast_free(entry->id);
92         entry->id = NULL;
93         ao2_cleanup(entry->snapshot);
94         entry->snapshot = NULL;
95 }
96
97 static struct cache_entry *cache_entry_create(struct stasis_message_type *type, const char *id, struct stasis_message *snapshot)
98 {
99         RAII_VAR(struct cache_entry *, entry, NULL, ao2_cleanup);
100
101         ast_assert(type != NULL);
102         ast_assert(id != NULL);
103
104         entry = ao2_alloc(sizeof(*entry), cache_entry_dtor);
105         if (!entry) {
106                 return NULL;
107         }
108
109         entry->id = ast_strdup(id);
110         if (!entry->id) {
111                 return NULL;
112         }
113
114         ao2_ref(type, +1);
115         entry->type = type;
116         if (snapshot != NULL) {
117                 ao2_ref(snapshot, +1);
118                 entry->snapshot = snapshot;
119         }
120
121         ao2_ref(entry, +1);
122         return entry;
123 }
124
125 static int cache_entry_hash(const void *obj, int flags)
126 {
127         const struct cache_entry *entry = obj;
128         int hash = 0;
129
130         ast_assert(!(flags & OBJ_KEY));
131
132         hash += ast_hashtab_hash_string(stasis_message_type_name(entry->type));
133         hash += ast_hashtab_hash_string(entry->id);
134         return hash;
135 }
136
137 static int cache_entry_cmp(void *obj, void *arg, int flags)
138 {
139         const struct cache_entry *left = obj;
140         const struct cache_entry *right = arg;
141
142         ast_assert(!(flags & OBJ_KEY));
143
144         if (left->type == right->type && strcmp(left->id, right->id) == 0) {
145                 return CMP_MATCH | CMP_STOP;
146         }
147
148         return 0;
149 }
150
151 static struct stasis_message *cache_put(struct stasis_caching_topic *caching_topic, struct stasis_message_type *type, const char *id, struct stasis_message *new_snapshot)
152 {
153         RAII_VAR(struct cache_entry *, new_entry, NULL, ao2_cleanup);
154         RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
155         struct stasis_message *old_snapshot = NULL;
156
157         ast_assert(caching_topic->cache != NULL);
158
159         new_entry = cache_entry_create(type, id, new_snapshot);
160
161         if (new_snapshot == NULL) {
162                 /* Remove entry from cache */
163                 cached_entry = ao2_find(caching_topic->cache, new_entry, OBJ_POINTER | OBJ_UNLINK);
164                 if (cached_entry) {
165                         old_snapshot = cached_entry->snapshot;
166                         cached_entry->snapshot = NULL;
167                 }
168         } else {
169                 /* Insert/update cache */
170                 SCOPED_AO2LOCK(lock, caching_topic->cache);
171
172                 cached_entry = ao2_find(caching_topic->cache, new_entry, OBJ_POINTER | OBJ_NOLOCK);
173                 if (cached_entry) {
174                         /* Update cache. Because objects are moving, no need to update refcounts. */
175                         old_snapshot = cached_entry->snapshot;
176                         cached_entry->snapshot = new_entry->snapshot;
177                         new_entry->snapshot = NULL;
178                 } else {
179                         /* Insert into the cache */
180                         ao2_link_flags(caching_topic->cache, new_entry, OBJ_NOLOCK);
181                 }
182
183         }
184
185         return old_snapshot;
186 }
187
188 struct stasis_message *stasis_cache_get(struct stasis_caching_topic *caching_topic, struct stasis_message_type *type, const char *id)
189 {
190         RAII_VAR(struct cache_entry *, search_entry, NULL, ao2_cleanup);
191         RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
192
193         ast_assert(caching_topic->cache != NULL);
194
195         search_entry = cache_entry_create(type, id, NULL);
196         if (search_entry == NULL) {
197                 return NULL;
198         }
199
200         cached_entry = ao2_find(caching_topic->cache, search_entry, OBJ_POINTER);
201         if (cached_entry == NULL) {
202                 return NULL;
203         }
204
205         ast_assert(cached_entry->snapshot != NULL);
206         ao2_ref(cached_entry->snapshot, +1);
207         return cached_entry->snapshot;
208 }
209
210 struct cache_dump_data {
211         struct ao2_container *cached;
212         struct stasis_message_type *type;
213 };
214
215 static int cache_dump_cb(void *obj, void *arg, int flags)
216 {
217         struct cache_dump_data *cache_dump = arg;
218         struct cache_entry *entry = obj;
219
220         if (!cache_dump->type || entry->type == cache_dump->type) {
221                 ao2_link(cache_dump->cached, entry->snapshot);
222         }
223
224         return 0;
225 }
226
227 struct ao2_container *stasis_cache_dump(struct stasis_caching_topic *caching_topic, struct stasis_message_type *type)
228 {
229         struct cache_dump_data cache_dump;
230
231         ast_assert(caching_topic->cache != NULL);
232
233         cache_dump.type = type;
234         cache_dump.cached = ao2_container_alloc(1, NULL, NULL);
235         if (!cache_dump.cached) {
236                 return NULL;
237         }
238
239         ao2_callback(caching_topic->cache, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_cb, &cache_dump);
240         return cache_dump.cached;
241 }
242
243 static struct stasis_message_type *__cache_clear_data;
244
245 static struct stasis_message_type *cache_clear_data(void)
246 {
247         ast_assert(__cache_clear_data != NULL);
248         return __cache_clear_data;
249 }
250
251 static struct stasis_message_type *__cache_update;
252
253 struct stasis_message_type *stasis_cache_update(void)
254 {
255         ast_assert(__cache_update != NULL);
256         return __cache_update;
257 }
258
259 struct cache_clear_data {
260         struct stasis_message_type *type;
261         char *id;
262 };
263
264 static void cache_clear_data_dtor(void *obj)
265 {
266         struct cache_clear_data *ev = obj;
267         ast_free(ev->id);
268         ev->id = NULL;
269         ao2_cleanup(ev->type);
270         ev->type = NULL;
271 }
272
273 struct stasis_message *stasis_cache_clear_create(struct stasis_message_type *type, const char *id)
274 {
275         RAII_VAR(struct cache_clear_data *, ev, NULL, ao2_cleanup);
276         RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
277
278         ev = ao2_alloc(sizeof(*ev), cache_clear_data_dtor);
279         if (!ev) {
280                 return NULL;
281         }
282
283         ev->id = ast_strdup(id);
284         if (!ev->id) {
285                 return NULL;
286         }
287         ao2_ref(type, +1);
288         ev->type = type;
289
290         msg = stasis_message_create(cache_clear_data(), ev);
291
292         if (!msg) {
293                 return NULL;
294         }
295
296         ao2_ref(msg, +1);
297         return msg;
298 }
299
300 static void stasis_cache_update_dtor(void *obj)
301 {
302         struct stasis_cache_update *update = obj;
303         ao2_cleanup(update->topic);
304         update->topic = NULL;
305         ao2_cleanup(update->old_snapshot);
306         update->old_snapshot = NULL;
307         ao2_cleanup(update->new_snapshot);
308         update->new_snapshot = NULL;
309         ao2_cleanup(update->type);
310         update->type = NULL;
311 }
312
313 static struct stasis_message *update_create(struct stasis_topic *topic, struct stasis_message *old_snapshot, struct stasis_message *new_snapshot)
314 {
315         RAII_VAR(struct stasis_cache_update *, update, NULL, ao2_cleanup);
316         RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
317
318         ast_assert(topic != NULL);
319         ast_assert(old_snapshot != NULL || new_snapshot != NULL);
320
321         update = ao2_alloc(sizeof(*update), stasis_cache_update_dtor);
322         if (!update) {
323                 return NULL;
324         }
325
326         ao2_ref(topic, +1);
327         update->topic = topic;
328         if (old_snapshot) {
329                 ao2_ref(old_snapshot, +1);
330                 update->old_snapshot = old_snapshot;
331                 if (!new_snapshot) {
332                         ao2_ref(stasis_message_type(old_snapshot), +1);
333                         update->type = stasis_message_type(old_snapshot);
334                 }
335         }
336         if (new_snapshot) {
337                 ao2_ref(new_snapshot, +1);
338                 update->new_snapshot = new_snapshot;
339                 ao2_ref(stasis_message_type(new_snapshot), +1);
340                 update->type = stasis_message_type(new_snapshot);
341         }
342
343         msg = stasis_message_create(stasis_cache_update(), update);
344         if (!msg) {
345                 return NULL;
346         }
347
348         ao2_ref(msg, +1);
349         return msg;
350 }
351
352 static void caching_topic_exec(void *data, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *message)
353 {
354         RAII_VAR(struct stasis_caching_topic *, caching_topic_needs_unref, NULL, ao2_cleanup);
355         struct stasis_caching_topic *caching_topic = data;
356         const char *id = NULL;
357
358         ast_assert(caching_topic->topic != NULL);
359         ast_assert(caching_topic->id_fn != NULL);
360
361         if (stasis_subscription_final_message(sub, message)) {
362                 caching_topic_needs_unref = caching_topic;
363         }
364
365         /* Handle cache clear event */
366         if (cache_clear_data() == stasis_message_type(message)) {
367                 RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
368                 RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
369                 struct cache_clear_data *clear = stasis_message_data(message);
370                 ast_assert(clear->type != NULL);
371                 ast_assert(clear->id != NULL);
372                 old_snapshot = cache_put(caching_topic, clear->type, clear->id, NULL);
373                 if (old_snapshot) {
374                         update = update_create(topic, old_snapshot, NULL);
375                         stasis_publish(caching_topic->topic, update);
376                 } else {
377                         ast_log(LOG_ERROR,
378                                 "Attempting to remove an item from the cache that isn't there: %s %s\n",
379                                 stasis_message_type_name(clear->type), clear->id);
380                 }
381                 return;
382         }
383
384         id = caching_topic->id_fn(message);
385         if (id == NULL) {
386                 /* Object isn't cached; forward */
387                 stasis_forward_message(caching_topic->topic, topic, message);
388         } else {
389                 /* Update the cache */
390                 RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
391                 RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
392
393                 old_snapshot = cache_put(caching_topic, stasis_message_type(message), id, message);
394
395                 update = update_create(topic, old_snapshot, message);
396                 if (update == NULL) {
397                         return;
398                 }
399
400                 stasis_publish(caching_topic->topic, update);
401         }
402
403         if (stasis_subscription_final_message(sub, message)) {
404                 ao2_cleanup(caching_topic);
405         }
406 }
407
408 struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *original_topic, snapshot_get_id id_fn)
409 {
410         RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, ao2_cleanup);
411         struct stasis_subscription *sub;
412         RAII_VAR(char *, new_name, NULL, free);
413         int ret;
414
415         ret = asprintf(&new_name, "%s-cached", stasis_topic_name(original_topic));
416         if (ret < 0) {
417                 return NULL;
418         }
419
420         caching_topic = ao2_alloc(sizeof(*caching_topic), stasis_caching_topic_dtor);
421         if (caching_topic == NULL) {
422                 return NULL;
423         }
424
425         caching_topic->cache = ao2_container_alloc(NUM_CACHE_BUCKETS, cache_entry_hash, cache_entry_cmp);
426         if (!caching_topic->cache) {
427                 ast_log(LOG_ERROR, "Stasis cache allocation failed\n");
428                 return NULL;
429         }
430
431         caching_topic->topic = stasis_topic_create(new_name);
432         if (caching_topic->topic == NULL) {
433                 return NULL;
434         }
435
436         caching_topic->id_fn = id_fn;
437
438         sub = stasis_subscribe(original_topic, caching_topic_exec, caching_topic);
439         if (sub == NULL) {
440                 return NULL;
441         }
442         /* This is for the reference contained in the subscription above */
443         ao2_ref(caching_topic, +1);
444         caching_topic->sub = sub;
445
446         ao2_ref(caching_topic, +1);
447         return caching_topic;
448 }
449
450 static void stasis_cache_exit(void)
451 {
452         ao2_cleanup(__cache_clear_data);
453         __cache_clear_data = NULL;
454         ao2_cleanup(__cache_update);
455         __cache_update = NULL;
456 }
457
458 int stasis_cache_init(void)
459 {
460         ast_register_atexit(stasis_cache_exit);
461
462         if (__cache_clear_data || __cache_update) {
463                 ast_log(LOG_ERROR, "Stasis cache double initialized\n");
464                 return -1;
465         }
466
467         __cache_update = stasis_message_type_create("stasis_cache_update");
468         if (!__cache_update) {
469                 return -1;
470         }
471
472         __cache_clear_data = stasis_message_type_create("StasisCacheClear");
473         if (!__cache_clear_data) {
474                 return -1;
475         }
476         return 0;
477 }
478