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->old_snapshot);
 343         update->old_snapshot = NULL;
 344         ao2_cleanup(update->new_snapshot);
 345         update->new_snapshot = NULL;
 346         ao2_cleanup(update->type);
 347         update->type = NULL;
 348 }
 349
 350 static struct stasis_message *update_create(struct stasis_message *old_snapshot, struct stasis_message *new_snapshot)
 351 {
 352         RAII_VAR(struct stasis_cache_update *, update, NULL, ao2_cleanup);
 353         RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
 354
 355         ast_assert(old_snapshot != NULL || new_snapshot != NULL);
 356
 357         update = ao2_alloc_options(sizeof(*update), stasis_cache_update_dtor,
 358                 AO2_ALLOC_OPT_LOCK_NOLOCK);
 359         if (!update) {
 360                 return NULL;
 361         }
 362
 363         if (old_snapshot) {
 364                 ao2_ref(old_snapshot, +1);
 365                 update->old_snapshot = old_snapshot;
 366                 if (!new_snapshot) {
 367                         ao2_ref(stasis_message_type(old_snapshot), +1);
 368                         update->type = stasis_message_type(old_snapshot);
 369                 }
 370         }
 371         if (new_snapshot) {
 372                 ao2_ref(new_snapshot, +1);
 373                 update->new_snapshot = new_snapshot;
 374                 ao2_ref(stasis_message_type(new_snapshot), +1);
 375                 update->type = stasis_message_type(new_snapshot);
 376         }
 377
 378         msg = stasis_message_create(stasis_cache_update_type(), update);
 379         if (!msg) {
 380                 return NULL;
 381         }
 382
 383         ao2_ref(msg, +1);
 384         return msg;
 385 }
 386
 387 static void caching_topic_exec(void *data, struct stasis_subscription *sub,
 388         struct stasis_message *message)
 389 {
 390         RAII_VAR(struct stasis_caching_topic *, caching_topic_needs_unref, NULL, ao2_cleanup);
 391         struct stasis_caching_topic *caching_topic = data;
 392         const char *id = NULL;
 393
 394         ast_assert(caching_topic != NULL);
 395         ast_assert(caching_topic->topic != NULL);
 396         ast_assert(caching_topic->cache != NULL);
 397         ast_assert(caching_topic->cache->id_fn != NULL);
 398
 399         if (stasis_subscription_final_message(sub, message)) {
 400                 caching_topic_needs_unref = caching_topic;
 401         }
 402
 403         /* Handle cache clear event */
 404         if (stasis_cache_clear_type() == stasis_message_type(message)) {
 405                 RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
 406                 RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
 407                 struct stasis_message *clear_msg = stasis_message_data(message);
 408                 const char *clear_id = caching_topic->cache->id_fn(clear_msg);
 409                 struct stasis_message_type *clear_type = stasis_message_type(clear_msg);
 410
 411                 ast_assert(clear_type != NULL);
 412
 413                 if (clear_id) {
 414                         old_snapshot = cache_put(caching_topic->cache, clear_type, clear_id, NULL);
 415                         if (old_snapshot) {
 416                                 update = update_create(old_snapshot, NULL);
 417                                 stasis_publish(caching_topic->topic, update);
 418                                 return;
 419                         }
 420
 421                         ast_log(LOG_ERROR,
 422                                 "Attempting to remove an item from the %s cache that isn't there: %s %s\n",
 423                                 stasis_topic_name(caching_topic->topic), stasis_message_type_name(clear_type), clear_id);
 424                         return;
 425                 }
 426         }
 427
 428         id = caching_topic->cache->id_fn(message);
 429         if (id == NULL) {
 430                 /* Object isn't cached; discard */
 431         } else {
 432                 /* Update the cache */
 433                 RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
 434                 RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
 435
 436                 old_snapshot = cache_put(caching_topic->cache, stasis_message_type(message), id, message);
 437
 438                 update = update_create(old_snapshot, message);
 439                 if (update == NULL) {
 440                         return;
 441                 }
 442
 443                 stasis_publish(caching_topic->topic, update);
 444         }
 445 }
 446
 447 struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *original_topic, struct stasis_cache *cache)
 448 {
 449         RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, ao2_cleanup);
 450         struct stasis_subscription *sub;
 451         RAII_VAR(char *, new_name, NULL, ast_free);
 452         int ret;
 453
 454         ret = ast_asprintf(&new_name, "%s-cached", stasis_topic_name(original_topic));
 455         if (ret < 0) {
 456                 return NULL;
 457         }
 458
 459         caching_topic = ao2_alloc_options(sizeof(*caching_topic),
 460                 stasis_caching_topic_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
 461         if (caching_topic == NULL) {
 462                 return NULL;
 463         }
 464
 465         caching_topic->topic = stasis_topic_create(new_name);
 466         if (caching_topic->topic == NULL) {
 467                 return NULL;
 468         }
 469
 470         ao2_ref(cache, +1);
 471         caching_topic->cache = cache;
 472
 473         sub = internal_stasis_subscribe(original_topic, caching_topic_exec, caching_topic, 0);
 474         if (sub == NULL) {
 475                 return NULL;
 476         }
 477
 478         ao2_ref(original_topic, +1);
 479         caching_topic->original_topic = original_topic;
 480
 481         /* This is for the reference contained in the subscription above */
 482         ao2_ref(caching_topic, +1);
 483         caching_topic->sub = sub;
 484
 485         /* The subscription holds the reference, so no additional ref bump. */
 486         return caching_topic;
 487 }
 488
 489 static void stasis_cache_cleanup(void)
 490 {
 491         STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_clear_type);
 492         STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_update_type);
 493 }
 494
 495 int stasis_cache_init(void)
 496 {
 497         ast_register_cleanup(stasis_cache_cleanup);
 498
 499         if (STASIS_MESSAGE_TYPE_INIT(stasis_cache_clear_type) != 0) {
 500                 return -1;
 501         }
 502
 503         if (STASIS_MESSAGE_TYPE_INIT(stasis_cache_update_type) != 0) {
 504                 return -1;
 505         }
 506
 507         return 0;
 508 }
 509