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