ARI: Add ability to raise arbitrary User Events
[asterisk/asterisk.git] / main / taskprocessor.c
1 /*
2  * Asterisk -- An open source telephony toolkit.
3  *
4  * Copyright (C) 2007-2013, Digium, Inc.
5  *
6  * Dwayne M. Hubbard <dhubbard@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 /*!
20  * \file
21  * \brief Maintain a container of uniquely-named taskprocessor threads that can be shared across modules.
22  *
23  * \author Dwayne Hubbard <dhubbard@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/_private.h"
35 #include "asterisk/module.h"
36 #include "asterisk/time.h"
37 #include "asterisk/astobj2.h"
38 #include "asterisk/cli.h"
39 #include "asterisk/taskprocessor.h"
40 #include "asterisk/sem.h"
41
42 /*!
43  * \brief tps_task structure is queued to a taskprocessor
44  *
45  * tps_tasks are processed in FIFO order and freed by the taskprocessing
46  * thread after the task handler returns.  The callback function that is assigned
47  * to the execute() function pointer is responsible for releasing datap resources if necessary.
48  */
49 struct tps_task {
50         /*! \brief The execute() task callback function pointer */
51         union {
52                 int (*execute)(void *datap);
53                 int (*execute_local)(struct ast_taskprocessor_local *local);
54         } callback;
55         /*! \brief The data pointer for the task execute() function */
56         void *datap;
57         /*! \brief AST_LIST_ENTRY overhead */
58         AST_LIST_ENTRY(tps_task) list;
59         unsigned int wants_local:1;
60 };
61
62 /*! \brief tps_taskprocessor_stats maintain statistics for a taskprocessor. */
63 struct tps_taskprocessor_stats {
64         /*! \brief This is the maximum number of tasks queued at any one time */
65         unsigned long max_qsize;
66         /*! \brief This is the current number of tasks processed */
67         unsigned long _tasks_processed_count;
68 };
69
70 /*! \brief A ast_taskprocessor structure is a singleton by name */
71 struct ast_taskprocessor {
72         /*! \brief Friendly name of the taskprocessor */
73         const char *name;
74         /*! \brief Taskprocessor statistics */
75         struct tps_taskprocessor_stats *stats;
76         void *local_data;
77         /*! \brief Taskprocessor current queue size */
78         long tps_queue_size;
79         /*! \brief Taskprocessor queue */
80         AST_LIST_HEAD_NOLOCK(tps_queue, tps_task) tps_queue;
81         /*! \brief Taskprocessor singleton list entry */
82         AST_LIST_ENTRY(ast_taskprocessor) list;
83         struct ast_taskprocessor_listener *listener;
84         /*! Indicates if the taskprocessor is currently executing a task */
85         unsigned int executing:1;
86 };
87
88 /*!
89  * \brief A listener for taskprocessors
90  *
91  * \since 12.0.0
92  *
93  * When a taskprocessor's state changes, the listener
94  * is notified of the change. This allows for tasks
95  * to be addressed in whatever way is appropriate for
96  * the module using the taskprocessor.
97  */
98 struct ast_taskprocessor_listener {
99         /*! The callbacks the taskprocessor calls into to notify of state changes */
100         const struct ast_taskprocessor_listener_callbacks *callbacks;
101         /*! The taskprocessor that the listener is listening to */
102         struct ast_taskprocessor *tps;
103         /*! Data private to the listener */
104         void *user_data;
105 };
106
107 #define TPS_MAX_BUCKETS 7
108 /*! \brief tps_singletons is the astobj2 container for taskprocessor singletons */
109 static struct ao2_container *tps_singletons;
110
111 /*! \brief CLI <example>taskprocessor ping &lt;blah&gt;</example> operation requires a ping condition */
112 static ast_cond_t cli_ping_cond;
113
114 /*! \brief CLI <example>taskprocessor ping &lt;blah&gt;</example> operation requires a ping condition lock */
115 AST_MUTEX_DEFINE_STATIC(cli_ping_cond_lock);
116
117 /*! \brief The astobj2 hash callback for taskprocessors */
118 static int tps_hash_cb(const void *obj, const int flags);
119 /*! \brief The astobj2 compare callback for taskprocessors */
120 static int tps_cmp_cb(void *obj, void *arg, int flags);
121
122 /*! \brief Destroy the taskprocessor when its refcount reaches zero */
123 static void tps_taskprocessor_destroy(void *tps);
124
125 /*! \brief CLI <example>taskprocessor ping &lt;blah&gt;</example> handler function */
126 static int tps_ping_handler(void *datap);
127
128 /*! \brief Remove the front task off the taskprocessor queue */
129 static struct tps_task *tps_taskprocessor_pop(struct ast_taskprocessor *tps);
130
131 /*! \brief Return the size of the taskprocessor queue */
132 static int tps_taskprocessor_depth(struct ast_taskprocessor *tps);
133
134 static char *cli_tps_ping(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a);
135 static char *cli_tps_report(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a);
136
137 static struct ast_cli_entry taskprocessor_clis[] = {
138         AST_CLI_DEFINE(cli_tps_ping, "Ping a named task processor"),
139         AST_CLI_DEFINE(cli_tps_report, "List instantiated task processors and statistics"),
140 };
141
142 struct default_taskprocessor_listener_pvt {
143         pthread_t poll_thread;
144         int dead;
145         struct ast_sem sem;
146 };
147
148 static void default_listener_pvt_destroy(struct default_taskprocessor_listener_pvt *pvt)
149 {
150         ast_assert(pvt->dead);
151         ast_sem_destroy(&pvt->sem);
152         ast_free(pvt);
153 }
154
155 static void default_listener_pvt_dtor(struct ast_taskprocessor_listener *listener)
156 {
157         struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
158
159         default_listener_pvt_destroy(pvt);
160
161         listener->user_data = NULL;
162 }
163
164 /*!
165  * \brief Function that processes tasks in the taskprocessor
166  * \internal
167  */
168 static void *default_tps_processing_function(void *data)
169 {
170         struct ast_taskprocessor_listener *listener = data;
171         struct ast_taskprocessor *tps = listener->tps;
172         struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
173         int sem_value;
174         int res;
175
176         while (!pvt->dead) {
177                 res = ast_sem_wait(&pvt->sem);
178                 if (res != 0 && errno != EINTR) {
179                         ast_log(LOG_ERROR, "ast_sem_wait(): %s\n",
180                                 strerror(errno));
181                         /* Just give up */
182                         break;
183                 }
184                 ast_taskprocessor_execute(tps);
185         }
186
187         /* No posting to a dead taskprocessor! */
188         res = ast_sem_getvalue(&pvt->sem, &sem_value);
189         ast_assert(res == 0 && sem_value == 0);
190
191         /* Free the shutdown reference (see default_listener_shutdown) */
192         ao2_t_ref(listener->tps, -1, "tps-shutdown");
193
194         return NULL;
195 }
196
197 static int default_listener_start(struct ast_taskprocessor_listener *listener)
198 {
199         struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
200
201         if (ast_pthread_create(&pvt->poll_thread, NULL, default_tps_processing_function, listener)) {
202                 return -1;
203         }
204
205         return 0;
206 }
207
208 static void default_task_pushed(struct ast_taskprocessor_listener *listener, int was_empty)
209 {
210         struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
211
212         if (ast_sem_post(&pvt->sem) != 0) {
213                 ast_log(LOG_ERROR, "Failed to notify of enqueued task: %s\n",
214                         strerror(errno));
215         }
216 }
217
218 static int default_listener_die(void *data)
219 {
220         struct default_taskprocessor_listener_pvt *pvt = data;
221         pvt->dead = 1;
222         return 0;
223 }
224
225 static void default_listener_shutdown(struct ast_taskprocessor_listener *listener)
226 {
227         struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
228         int res;
229
230         /* Hold a reference during shutdown */
231         ao2_t_ref(listener->tps, +1, "tps-shutdown");
232
233         ast_taskprocessor_push(listener->tps, default_listener_die, pvt);
234
235         ast_assert(pvt->poll_thread != AST_PTHREADT_NULL);
236
237         if (pthread_equal(pthread_self(), pvt->poll_thread)) {
238                 res = pthread_detach(pvt->poll_thread);
239                 if (res != 0) {
240                         ast_log(LOG_ERROR, "pthread_detach(): %s\n", strerror(errno));
241                 }
242         } else {
243                 res = pthread_join(pvt->poll_thread, NULL);
244                 if (res != 0) {
245                         ast_log(LOG_ERROR, "pthread_join(): %s\n", strerror(errno));
246                 }
247         }
248         pvt->poll_thread = AST_PTHREADT_NULL;
249 }
250
251 static const struct ast_taskprocessor_listener_callbacks default_listener_callbacks = {
252         .start = default_listener_start,
253         .task_pushed = default_task_pushed,
254         .shutdown = default_listener_shutdown,
255         .dtor = default_listener_pvt_dtor,
256 };
257
258 /*!
259  * \internal
260  * \brief Clean up resources on Asterisk shutdown
261  */
262 static void tps_shutdown(void)
263 {
264         ast_cli_unregister_multiple(taskprocessor_clis, ARRAY_LEN(taskprocessor_clis));
265         ao2_t_ref(tps_singletons, -1, "Unref tps_singletons in shutdown");
266         tps_singletons = NULL;
267 }
268
269 /* initialize the taskprocessor container and register CLI operations */
270 int ast_tps_init(void)
271 {
272         if (!(tps_singletons = ao2_container_alloc(TPS_MAX_BUCKETS, tps_hash_cb, tps_cmp_cb))) {
273                 ast_log(LOG_ERROR, "taskprocessor container failed to initialize!\n");
274                 return -1;
275         }
276
277         ast_cond_init(&cli_ping_cond, NULL);
278
279         ast_cli_register_multiple(taskprocessor_clis, ARRAY_LEN(taskprocessor_clis));
280
281         ast_register_atexit(tps_shutdown);
282
283         return 0;
284 }
285
286 /* allocate resources for the task */
287 static struct tps_task *tps_task_alloc(int (*task_exe)(void *datap), void *datap)
288 {
289         struct tps_task *t;
290         if (!task_exe) {
291                 ast_log(LOG_ERROR, "task_exe is NULL!\n");
292                 return NULL;
293         }
294
295         t = ast_calloc(1, sizeof(*t));
296         if (!t) {
297                 ast_log(LOG_ERROR, "failed to allocate task!\n");
298                 return NULL;
299         }
300
301         t->callback.execute = task_exe;
302         t->datap = datap;
303
304         return t;
305 }
306
307 static struct tps_task *tps_task_alloc_local(int (*task_exe)(struct ast_taskprocessor_local *local), void *datap)
308 {
309         struct tps_task *t;
310         if (!task_exe) {
311                 ast_log(LOG_ERROR, "task_exe is NULL!\n");
312                 return NULL;
313         }
314
315         t = ast_calloc(1, sizeof(*t));
316         if (!t) {
317                 ast_log(LOG_ERROR, "failed to allocate task!\n");
318                 return NULL;
319         }
320
321         t->callback.execute_local = task_exe;
322         t->datap = datap;
323         t->wants_local = 1;
324
325         return t;
326 }
327
328 /* release task resources */
329 static void *tps_task_free(struct tps_task *task)
330 {
331         ast_free(task);
332         return NULL;
333 }
334
335 /* taskprocessor tab completion */
336 static char *tps_taskprocessor_tab_complete(struct ast_taskprocessor *p, struct ast_cli_args *a)
337 {
338         int tklen;
339         int wordnum = 0;
340         char *name = NULL;
341         struct ao2_iterator i;
342
343         if (a->pos != 3)
344                 return NULL;
345
346         tklen = strlen(a->word);
347         i = ao2_iterator_init(tps_singletons, 0);
348         while ((p = ao2_iterator_next(&i))) {
349                 if (!strncasecmp(a->word, p->name, tklen) && ++wordnum > a->n) {
350                         name = ast_strdup(p->name);
351                         ao2_ref(p, -1);
352                         break;
353                 }
354                 ao2_ref(p, -1);
355         }
356         ao2_iterator_destroy(&i);
357         return name;
358 }
359
360 /* ping task handling function */
361 static int tps_ping_handler(void *datap)
362 {
363         ast_mutex_lock(&cli_ping_cond_lock);
364         ast_cond_signal(&cli_ping_cond);
365         ast_mutex_unlock(&cli_ping_cond_lock);
366         return 0;
367 }
368
369 /* ping the specified taskprocessor and display the ping time on the CLI */
370 static char *cli_tps_ping(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
371 {
372         struct timeval begin, end, delta;
373         const char *name;
374         struct timeval when;
375         struct timespec ts;
376         struct ast_taskprocessor *tps = NULL;
377
378         switch (cmd) {
379         case CLI_INIT:
380                 e->command = "core ping taskprocessor";
381                 e->usage =
382                         "Usage: core ping taskprocessor <taskprocessor>\n"
383                         "       Displays the time required for a task to be processed\n";
384                 return NULL;
385         case CLI_GENERATE:
386                 return tps_taskprocessor_tab_complete(tps, a);
387         }
388
389         if (a->argc != 4)
390                 return CLI_SHOWUSAGE;
391
392         name = a->argv[3];
393         if (!(tps = ast_taskprocessor_get(name, TPS_REF_IF_EXISTS))) {
394                 ast_cli(a->fd, "\nping failed: %s not found\n\n", name);
395                 return CLI_SUCCESS;
396         }
397         ast_cli(a->fd, "\npinging %s ...", name);
398         when = ast_tvadd((begin = ast_tvnow()), ast_samp2tv(1000, 1000));
399         ts.tv_sec = when.tv_sec;
400         ts.tv_nsec = when.tv_usec * 1000;
401         ast_mutex_lock(&cli_ping_cond_lock);
402         if (ast_taskprocessor_push(tps, tps_ping_handler, 0) < 0) {
403                 ast_cli(a->fd, "\nping failed: could not push task to %s\n\n", name);
404                 ao2_ref(tps, -1);
405                 return CLI_FAILURE;
406         }
407         ast_cond_timedwait(&cli_ping_cond, &cli_ping_cond_lock, &ts);
408         ast_mutex_unlock(&cli_ping_cond_lock);
409         end = ast_tvnow();
410         delta = ast_tvsub(end, begin);
411         ast_cli(a->fd, "\n\t%24s ping time: %.1ld.%.6ld sec\n\n", name, (long)delta.tv_sec, (long int)delta.tv_usec);
412         ao2_ref(tps, -1);
413         return CLI_SUCCESS;
414 }
415
416 static char *cli_tps_report(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
417 {
418         char name[256];
419         int tcount;
420         unsigned long qsize;
421         unsigned long maxqsize;
422         unsigned long processed;
423         struct ast_taskprocessor *p;
424         struct ao2_iterator i;
425
426         switch (cmd) {
427         case CLI_INIT:
428                 e->command = "core show taskprocessors";
429                 e->usage =
430                         "Usage: core show taskprocessors\n"
431                         "       Shows a list of instantiated task processors and their statistics\n";
432                 return NULL;
433         case CLI_GENERATE:
434                 return NULL;
435         }
436
437         if (a->argc != e->args)
438                 return CLI_SHOWUSAGE;
439
440         ast_cli(a->fd, "\n\t+----- Processor -----+--- Processed ---+- In Queue -+- Max Depth -+");
441         i = ao2_iterator_init(tps_singletons, 0);
442         while ((p = ao2_iterator_next(&i))) {
443                 ast_copy_string(name, p->name, sizeof(name));
444                 qsize = p->tps_queue_size;
445                 maxqsize = p->stats->max_qsize;
446                 processed = p->stats->_tasks_processed_count;
447                 ast_cli(a->fd, "\n%24s   %17lu %12lu %12lu", name, processed, qsize, maxqsize);
448                 ao2_ref(p, -1);
449         }
450         ao2_iterator_destroy(&i);
451         tcount = ao2_container_count(tps_singletons);
452         ast_cli(a->fd, "\n\t+---------------------+-----------------+------------+-------------+\n\t%d taskprocessors\n\n", tcount);
453         return CLI_SUCCESS;
454 }
455
456 /* hash callback for astobj2 */
457 static int tps_hash_cb(const void *obj, const int flags)
458 {
459         const struct ast_taskprocessor *tps = obj;
460         const char *name = flags & OBJ_KEY ? obj : tps->name;
461
462         return ast_str_case_hash(name);
463 }
464
465 /* compare callback for astobj2 */
466 static int tps_cmp_cb(void *obj, void *arg, int flags)
467 {
468         struct ast_taskprocessor *lhs = obj, *rhs = arg;
469         const char *rhsname = flags & OBJ_KEY ? arg : rhs->name;
470
471         return !strcasecmp(lhs->name, rhsname) ? CMP_MATCH | CMP_STOP : 0;
472 }
473
474 /* destroy the taskprocessor */
475 static void tps_taskprocessor_destroy(void *tps)
476 {
477         struct ast_taskprocessor *t = tps;
478         struct tps_task *task;
479
480         if (!tps) {
481                 ast_log(LOG_ERROR, "missing taskprocessor\n");
482                 return;
483         }
484         ast_debug(1, "destroying taskprocessor '%s'\n", t->name);
485         /* free it */
486         ast_free(t->stats);
487         t->stats = NULL;
488         ast_free((char *) t->name);
489         if (t->listener) {
490                 ao2_ref(t->listener, -1);
491                 t->listener = NULL;
492         }
493         while ((task = AST_LIST_REMOVE_HEAD(&t->tps_queue, list))) {
494                 tps_task_free(task);
495         }
496 }
497
498 /* pop the front task and return it */
499 static struct tps_task *tps_taskprocessor_pop(struct ast_taskprocessor *tps)
500 {
501         struct tps_task *task;
502
503         if ((task = AST_LIST_REMOVE_HEAD(&tps->tps_queue, list))) {
504                 tps->tps_queue_size--;
505         }
506         return task;
507 }
508
509 static int tps_taskprocessor_depth(struct ast_taskprocessor *tps)
510 {
511         return (tps) ? tps->tps_queue_size : -1;
512 }
513
514 /* taskprocessor name accessor */
515 const char *ast_taskprocessor_name(struct ast_taskprocessor *tps)
516 {
517         if (!tps) {
518                 ast_log(LOG_ERROR, "no taskprocessor specified!\n");
519                 return NULL;
520         }
521         return tps->name;
522 }
523
524 static void listener_shutdown(struct ast_taskprocessor_listener *listener)
525 {
526         listener->callbacks->shutdown(listener);
527         ao2_ref(listener->tps, -1);
528 }
529
530 static void taskprocessor_listener_dtor(void *obj)
531 {
532         struct ast_taskprocessor_listener *listener = obj;
533
534         if (listener->callbacks->dtor) {
535                 listener->callbacks->dtor(listener);
536         }
537 }
538
539 struct ast_taskprocessor_listener *ast_taskprocessor_listener_alloc(const struct ast_taskprocessor_listener_callbacks *callbacks, void *user_data)
540 {
541         struct ast_taskprocessor_listener *listener;
542
543         listener = ao2_alloc(sizeof(*listener), taskprocessor_listener_dtor);
544         if (!listener) {
545                 return NULL;
546         }
547         listener->callbacks = callbacks;
548         listener->user_data = user_data;
549
550         return listener;
551 }
552
553 struct ast_taskprocessor *ast_taskprocessor_listener_get_tps(const struct ast_taskprocessor_listener *listener)
554 {
555         ao2_ref(listener->tps, +1);
556         return listener->tps;
557 }
558
559 void *ast_taskprocessor_listener_get_user_data(const struct ast_taskprocessor_listener *listener)
560 {
561         return listener->user_data;
562 }
563
564 static void *default_listener_pvt_alloc(void)
565 {
566         struct default_taskprocessor_listener_pvt *pvt;
567
568         pvt = ast_calloc(1, sizeof(*pvt));
569         if (!pvt) {
570                 return NULL;
571         }
572         pvt->poll_thread = AST_PTHREADT_NULL;
573         if (ast_sem_init(&pvt->sem, 0, 0) != 0) {
574                 ast_log(LOG_ERROR, "ast_sem_init(): %s\n", strerror(errno));
575                 ast_free(pvt);
576                 return NULL;
577         }
578         return pvt;
579 }
580
581 static struct ast_taskprocessor *__allocate_taskprocessor(const char *name, struct ast_taskprocessor_listener *listener)
582 {
583         RAII_VAR(struct ast_taskprocessor *, p,
584                         ao2_alloc(sizeof(*p), tps_taskprocessor_destroy), ao2_cleanup);
585
586         if (!p) {
587                 ast_log(LOG_WARNING, "failed to create taskprocessor '%s'\n", name);
588                 return NULL;
589         }
590
591         if (!(p->stats = ast_calloc(1, sizeof(*p->stats)))) {
592                 ast_log(LOG_WARNING, "failed to create taskprocessor stats for '%s'\n", name);
593                 return NULL;
594         }
595         if (!(p->name = ast_strdup(name))) {
596                 ao2_ref(p, -1);
597                 return NULL;
598         }
599
600         ao2_ref(listener, +1);
601         p->listener = listener;
602
603         ao2_ref(p, +1);
604         listener->tps = p;
605
606         if (!(ao2_link(tps_singletons, p))) {
607                 ast_log(LOG_ERROR, "Failed to add taskprocessor '%s' to container\n", p->name);
608                 return NULL;
609         }
610
611         if (p->listener->callbacks->start(p->listener)) {
612                 ast_log(LOG_ERROR, "Unable to start taskprocessor listener for taskprocessor %s\n", p->name);
613                 ast_taskprocessor_unreference(p);
614                 return NULL;
615         }
616
617         /* RAII_VAR will decrement the refcount at the end of the function.
618          * Since we want to pass back a reference to p, we bump the refcount
619          */
620         ao2_ref(p, +1);
621         return p;
622
623 }
624
625 /* Provide a reference to a taskprocessor.  Create the taskprocessor if necessary, but don't
626  * create the taskprocessor if we were told via ast_tps_options to return a reference only
627  * if it already exists */
628 struct ast_taskprocessor *ast_taskprocessor_get(const char *name, enum ast_tps_options create)
629 {
630         struct ast_taskprocessor *p;
631         struct ast_taskprocessor_listener *listener;
632         struct default_taskprocessor_listener_pvt *pvt;
633
634         if (ast_strlen_zero(name)) {
635                 ast_log(LOG_ERROR, "requesting a nameless taskprocessor!!!\n");
636                 return NULL;
637         }
638         p = ao2_find(tps_singletons, name, OBJ_KEY);
639         if (p) {
640                 return p;
641         }
642         if (create & TPS_REF_IF_EXISTS) {
643                 /* calling function does not want a new taskprocessor to be created if it doesn't already exist */
644                 return NULL;
645         }
646         /* Create a new taskprocessor. Start by creating a default listener */
647         pvt = default_listener_pvt_alloc();
648         if (!pvt) {
649                 return NULL;
650         }
651         listener = ast_taskprocessor_listener_alloc(&default_listener_callbacks, pvt);
652         if (!listener) {
653                 default_listener_pvt_destroy(pvt);
654                 return NULL;
655         }
656
657         p = __allocate_taskprocessor(name, listener);
658         if (!p) {
659                 ao2_ref(listener, -1);
660                 return NULL;
661         }
662
663         /* Unref listener here since the taskprocessor has gained a reference to the listener */
664         ao2_ref(listener, -1);
665         return p;
666 }
667
668 struct ast_taskprocessor *ast_taskprocessor_create_with_listener(const char *name, struct ast_taskprocessor_listener *listener)
669 {
670         struct ast_taskprocessor *p = ao2_find(tps_singletons, name, OBJ_KEY);
671
672         if (p) {
673                 ast_taskprocessor_unreference(p);
674                 return NULL;
675         }
676         return __allocate_taskprocessor(name, listener);
677 }
678
679 void ast_taskprocessor_set_local(struct ast_taskprocessor *tps,
680         void *local_data)
681 {
682         SCOPED_AO2LOCK(lock, tps);
683         tps->local_data = local_data;
684 }
685
686 /* decrement the taskprocessor reference count and unlink from the container if necessary */
687 void *ast_taskprocessor_unreference(struct ast_taskprocessor *tps)
688 {
689         if (!tps) {
690                 return NULL;
691         }
692
693         if (ao2_ref(tps, -1) > 3) {
694                 return NULL;
695         }
696         /* If we're down to 3 references, then those must be:
697          * 1. The reference we just got rid of
698          * 2. The container
699          * 3. The listener
700          */
701         ao2_unlink(tps_singletons, tps);
702         listener_shutdown(tps->listener);
703         return NULL;
704 }
705
706 /* push the task into the taskprocessor queue */
707 static int taskprocessor_push(struct ast_taskprocessor *tps, struct tps_task *t)
708 {
709         int previous_size;
710         int was_empty;
711
712         if (!tps) {
713                 ast_log(LOG_ERROR, "tps is NULL!\n");
714                 return -1;
715         }
716
717         if (!t) {
718                 ast_log(LOG_ERROR, "t is NULL!\n");
719                 return -1;
720         }
721
722         ao2_lock(tps);
723         AST_LIST_INSERT_TAIL(&tps->tps_queue, t, list);
724         previous_size = tps->tps_queue_size++;
725         /* The currently executing task counts as still in queue */
726         was_empty = tps->executing ? 0 : previous_size == 0;
727         ao2_unlock(tps);
728         tps->listener->callbacks->task_pushed(tps->listener, was_empty);
729         return 0;
730 }
731
732 int ast_taskprocessor_push(struct ast_taskprocessor *tps, int (*task_exe)(void *datap), void *datap)
733 {
734         return taskprocessor_push(tps, tps_task_alloc(task_exe, datap));
735 }
736
737 int ast_taskprocessor_push_local(struct ast_taskprocessor *tps, int (*task_exe)(struct ast_taskprocessor_local *datap), void *datap)
738 {
739         return taskprocessor_push(tps, tps_task_alloc_local(task_exe, datap));
740 }
741
742 int ast_taskprocessor_execute(struct ast_taskprocessor *tps)
743 {
744         struct ast_taskprocessor_local local;
745         struct tps_task *t;
746         int size;
747
748         ao2_lock(tps);
749         t = tps_taskprocessor_pop(tps);
750         if (!t) {
751                 ao2_unlock(tps);
752                 return 0;
753         }
754
755         tps->executing = 1;
756
757         if (t->wants_local) {
758                 local.local_data = tps->local_data;
759                 local.data = t->datap;
760         }
761         ao2_unlock(tps);
762
763         if (t->wants_local) {
764                 t->callback.execute_local(&local);
765         } else {
766                 t->callback.execute(t->datap);
767         }
768         tps_task_free(t);
769
770         ao2_lock(tps);
771         /* We need to check size in the same critical section where we reset the
772          * executing bit. Avoids a race condition where a task is pushed right
773          * after we pop an empty stack.
774          */
775         tps->executing = 0;
776         size = tps_taskprocessor_depth(tps);
777         /* If we executed a task, bump the stats */
778         if (tps->stats) {
779                 tps->stats->_tasks_processed_count++;
780                 if (size > tps->stats->max_qsize) {
781                         tps->stats->max_qsize = size;
782                 }
783         }
784         ao2_unlock(tps);
785
786         /* If we executed a task, check for the transition to empty */
787         if (size == 0 && tps->listener->callbacks->emptied) {
788                 tps->listener->callbacks->emptied(tps->listener);
789         }
790         return size > 0;
791 }