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