CHANNEL(callid): Give dialplan access to the callid.
[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 #include "asterisk/_private.h"
33 #include "asterisk/module.h"
34 #include "asterisk/time.h"
35 #include "asterisk/astobj2.h"
36 #include "asterisk/cli.h"
37 #include "asterisk/taskprocessor.h"
38 #include "asterisk/sem.h"
39
40 /*!
41  * \brief tps_task structure is queued to a taskprocessor
42  *
43  * tps_tasks are processed in FIFO order and freed by the taskprocessing
44  * thread after the task handler returns.  The callback function that is assigned
45  * to the execute() function pointer is responsible for releasing datap resources if necessary.
46  */
47 struct tps_task {
48         /*! \brief The execute() task callback function pointer */
49         union {
50                 int (*execute)(void *datap);
51                 int (*execute_local)(struct ast_taskprocessor_local *local);
52         } callback;
53         /*! \brief The data pointer for the task execute() function */
54         void *datap;
55         /*! \brief AST_LIST_ENTRY overhead */
56         AST_LIST_ENTRY(tps_task) list;
57         unsigned int wants_local:1;
58 };
59
60 /*! \brief tps_taskprocessor_stats maintain statistics for a taskprocessor. */
61 struct tps_taskprocessor_stats {
62         /*! \brief This is the maximum number of tasks queued at any one time */
63         unsigned long max_qsize;
64         /*! \brief This is the current number of tasks processed */
65         unsigned long _tasks_processed_count;
66 };
67
68 /*! \brief A ast_taskprocessor structure is a singleton by name */
69 struct ast_taskprocessor {
70         /*! \brief Friendly name of the taskprocessor */
71         const char *name;
72         /*! \brief Taskprocessor statistics */
73         struct tps_taskprocessor_stats *stats;
74         void *local_data;
75         /*! \brief Taskprocessor current queue size */
76         long tps_queue_size;
77         /*! \brief Taskprocessor low water clear alert level */
78         long tps_queue_low;
79         /*! \brief Taskprocessor high water alert trigger level */
80         long tps_queue_high;
81         /*! \brief Taskprocessor queue */
82         AST_LIST_HEAD_NOLOCK(tps_queue, tps_task) tps_queue;
83         struct ast_taskprocessor_listener *listener;
84         /*! Current thread executing the tasks */
85         pthread_t thread;
86         /*! Indicates if the taskprocessor is currently executing a task */
87         unsigned int executing:1;
88         /*! Indicates that a high water warning has been issued on this task processor */
89         unsigned int high_water_warned:1;
90         /*! Indicates that a high water alert is active on this taskprocessor */
91         unsigned int high_water_alert:1;
92         /*! Indicates if the taskprocessor is currently suspended */
93         unsigned int suspended:1;
94 };
95
96 /*!
97  * \brief A listener for taskprocessors
98  *
99  * \since 12.0.0
100  *
101  * When a taskprocessor's state changes, the listener
102  * is notified of the change. This allows for tasks
103  * to be addressed in whatever way is appropriate for
104  * the module using the taskprocessor.
105  */
106 struct ast_taskprocessor_listener {
107         /*! The callbacks the taskprocessor calls into to notify of state changes */
108         const struct ast_taskprocessor_listener_callbacks *callbacks;
109         /*! The taskprocessor that the listener is listening to */
110         struct ast_taskprocessor *tps;
111         /*! Data private to the listener */
112         void *user_data;
113 };
114
115 #define TPS_MAX_BUCKETS 7
116 /*! \brief tps_singletons is the astobj2 container for taskprocessor singletons */
117 static struct ao2_container *tps_singletons;
118
119 /*! \brief CLI <example>taskprocessor ping &lt;blah&gt;</example> operation requires a ping condition */
120 static ast_cond_t cli_ping_cond;
121
122 /*! \brief CLI <example>taskprocessor ping &lt;blah&gt;</example> operation requires a ping condition lock */
123 AST_MUTEX_DEFINE_STATIC(cli_ping_cond_lock);
124
125 /*! \brief The astobj2 hash callback for taskprocessors */
126 static int tps_hash_cb(const void *obj, const int flags);
127 /*! \brief The astobj2 compare callback for taskprocessors */
128 static int tps_cmp_cb(void *obj, void *arg, int flags);
129
130 /*! \brief CLI <example>taskprocessor ping &lt;blah&gt;</example> handler function */
131 static int tps_ping_handler(void *datap);
132
133 static char *cli_tps_ping(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a);
134 static char *cli_tps_report(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a);
135
136 static struct ast_cli_entry taskprocessor_clis[] = {
137         AST_CLI_DEFINE(cli_tps_ping, "Ping a named task processor"),
138         AST_CLI_DEFINE(cli_tps_report, "List instantiated task processors and statistics"),
139 };
140
141 struct default_taskprocessor_listener_pvt {
142         pthread_t poll_thread;
143         int dead;
144         struct ast_sem sem;
145 };
146
147 static void default_listener_pvt_destroy(struct default_taskprocessor_listener_pvt *pvt)
148 {
149         ast_assert(pvt->dead);
150         ast_sem_destroy(&pvt->sem);
151         ast_free(pvt);
152 }
153
154 static void default_listener_pvt_dtor(struct ast_taskprocessor_listener *listener)
155 {
156         struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
157
158         default_listener_pvt_destroy(pvt);
159
160         listener->user_data = NULL;
161 }
162
163 /*!
164  * \brief Function that processes tasks in the taskprocessor
165  * \internal
166  */
167 static void *default_tps_processing_function(void *data)
168 {
169         struct ast_taskprocessor_listener *listener = data;
170         struct ast_taskprocessor *tps = listener->tps;
171         struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
172         int sem_value;
173         int res;
174
175         while (!pvt->dead) {
176                 res = ast_sem_wait(&pvt->sem);
177                 if (res != 0 && errno != EINTR) {
178                         ast_log(LOG_ERROR, "ast_sem_wait(): %s\n",
179                                 strerror(errno));
180                         /* Just give up */
181                         break;
182                 }
183                 ast_taskprocessor_execute(tps);
184         }
185
186         /* No posting to a dead taskprocessor! */
187         res = ast_sem_getvalue(&pvt->sem, &sem_value);
188         ast_assert(res == 0 && sem_value == 0);
189
190         /* Free the shutdown reference (see default_listener_shutdown) */
191         ao2_t_ref(listener->tps, -1, "tps-shutdown");
192
193         return NULL;
194 }
195
196 static int default_listener_start(struct ast_taskprocessor_listener *listener)
197 {
198         struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
199
200         if (ast_pthread_create(&pvt->poll_thread, NULL, default_tps_processing_function, listener)) {
201                 return -1;
202         }
203
204         return 0;
205 }
206
207 static void default_task_pushed(struct ast_taskprocessor_listener *listener, int was_empty)
208 {
209         struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
210
211         if (ast_sem_post(&pvt->sem) != 0) {
212                 ast_log(LOG_ERROR, "Failed to notify of enqueued task: %s\n",
213                         strerror(errno));
214         }
215 }
216
217 static int default_listener_die(void *data)
218 {
219         struct default_taskprocessor_listener_pvt *pvt = data;
220         pvt->dead = 1;
221         return 0;
222 }
223
224 static void default_listener_shutdown(struct ast_taskprocessor_listener *listener)
225 {
226         struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
227         int res;
228
229         /* Hold a reference during shutdown */
230         ao2_t_ref(listener->tps, +1, "tps-shutdown");
231
232         ast_taskprocessor_push(listener->tps, default_listener_die, pvt);
233
234         ast_assert(pvt->poll_thread != AST_PTHREADT_NULL);
235
236         if (pthread_equal(pthread_self(), pvt->poll_thread)) {
237                 res = pthread_detach(pvt->poll_thread);
238                 if (res != 0) {
239                         ast_log(LOG_ERROR, "pthread_detach(): %s\n", strerror(errno));
240                 }
241         } else {
242                 res = pthread_join(pvt->poll_thread, NULL);
243                 if (res != 0) {
244                         ast_log(LOG_ERROR, "pthread_join(): %s\n", strerror(errno));
245                 }
246         }
247         pvt->poll_thread = AST_PTHREADT_NULL;
248 }
249
250 static const struct ast_taskprocessor_listener_callbacks default_listener_callbacks = {
251         .start = default_listener_start,
252         .task_pushed = default_task_pushed,
253         .shutdown = default_listener_shutdown,
254         .dtor = default_listener_pvt_dtor,
255 };
256
257 /*!
258  * \internal
259  * \brief Clean up resources on Asterisk shutdown
260  */
261 static void tps_shutdown(void)
262 {
263         ast_cli_unregister_multiple(taskprocessor_clis, ARRAY_LEN(taskprocessor_clis));
264         ao2_t_ref(tps_singletons, -1, "Unref tps_singletons in shutdown");
265         tps_singletons = NULL;
266 }
267
268 /* initialize the taskprocessor container and register CLI operations */
269 int ast_tps_init(void)
270 {
271         if (!(tps_singletons = ao2_container_alloc(TPS_MAX_BUCKETS, tps_hash_cb, tps_cmp_cb))) {
272                 ast_log(LOG_ERROR, "taskprocessor container failed to initialize!\n");
273                 return -1;
274         }
275
276         ast_cond_init(&cli_ping_cond, NULL);
277
278         ast_cli_register_multiple(taskprocessor_clis, ARRAY_LEN(taskprocessor_clis));
279
280         ast_register_cleanup(tps_shutdown);
281
282         return 0;
283 }
284
285 /* allocate resources for the task */
286 static struct tps_task *tps_task_alloc(int (*task_exe)(void *datap), void *datap)
287 {
288         struct tps_task *t;
289         if (!task_exe) {
290                 ast_log(LOG_ERROR, "task_exe is NULL!\n");
291                 return NULL;
292         }
293
294         t = ast_calloc(1, sizeof(*t));
295         if (!t) {
296                 ast_log(LOG_ERROR, "failed to allocate task!\n");
297                 return NULL;
298         }
299
300         t->callback.execute = task_exe;
301         t->datap = datap;
302
303         return t;
304 }
305
306 static struct tps_task *tps_task_alloc_local(int (*task_exe)(struct ast_taskprocessor_local *local), void *datap)
307 {
308         struct tps_task *t;
309         if (!task_exe) {
310                 ast_log(LOG_ERROR, "task_exe is NULL!\n");
311                 return NULL;
312         }
313
314         t = ast_calloc(1, sizeof(*t));
315         if (!t) {
316                 ast_log(LOG_ERROR, "failed to allocate task!\n");
317                 return NULL;
318         }
319
320         t->callback.execute_local = task_exe;
321         t->datap = datap;
322         t->wants_local = 1;
323
324         return t;
325 }
326
327 /* release task resources */
328 static void *tps_task_free(struct tps_task *task)
329 {
330         ast_free(task);
331         return NULL;
332 }
333
334 /* taskprocessor tab completion */
335 static char *tps_taskprocessor_tab_complete(struct ast_cli_args *a)
336 {
337         int tklen;
338         int wordnum = 0;
339         struct ast_taskprocessor *p;
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                         ast_taskprocessor_unreference(p);
352                         break;
353                 }
354                 ast_taskprocessor_unreference(p);
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;
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(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
399         /*
400          * Wait up to 5 seconds for a ping reply.
401          *
402          * On a very busy system it could take awhile to get a
403          * ping response from some taskprocessors.
404          */
405         begin = ast_tvnow();
406         when = ast_tvadd(begin, ast_samp2tv(5000, 1000));
407         ts.tv_sec = when.tv_sec;
408         ts.tv_nsec = when.tv_usec * 1000;
409
410         ast_mutex_lock(&cli_ping_cond_lock);
411         if (ast_taskprocessor_push(tps, tps_ping_handler, 0) < 0) {
412                 ast_mutex_unlock(&cli_ping_cond_lock);
413                 ast_cli(a->fd, "\nping failed: could not push task to %s\n\n", name);
414                 ast_taskprocessor_unreference(tps);
415                 return CLI_FAILURE;
416         }
417         ast_cond_timedwait(&cli_ping_cond, &cli_ping_cond_lock, &ts);
418         ast_mutex_unlock(&cli_ping_cond_lock);
419
420         end = ast_tvnow();
421         delta = ast_tvsub(end, begin);
422         ast_cli(a->fd, "\n\t%24s ping time: %.1ld.%.6ld sec\n\n", name, (long)delta.tv_sec, (long int)delta.tv_usec);
423         ast_taskprocessor_unreference(tps);
424         return CLI_SUCCESS;
425 }
426
427 /*!
428  * \internal
429  * \brief Taskprocessor ao2 container sort function.
430  * \since 13.8.0
431  *
432  * \param obj_left pointer to the (user-defined part) of an object.
433  * \param obj_right pointer to the (user-defined part) of an object.
434  * \param flags flags from ao2_callback()
435  *   OBJ_SEARCH_OBJECT - if set, 'obj_right', is an object.
436  *   OBJ_SEARCH_KEY - if set, 'obj_right', is a search key item that is not an object.
437  *   OBJ_SEARCH_PARTIAL_KEY - if set, 'obj_right', is a partial search key item that is not an object.
438  *
439  * \retval <0 if obj_left < obj_right
440  * \retval =0 if obj_left == obj_right
441  * \retval >0 if obj_left > obj_right
442  */
443 static int tps_sort_cb(const void *obj_left, const void *obj_right, int flags)
444 {
445         const struct ast_taskprocessor *tps_left = obj_left;
446         const struct ast_taskprocessor *tps_right = obj_right;
447         const char *right_key = obj_right;
448         int cmp;
449
450         switch (flags & OBJ_SEARCH_MASK) {
451         default:
452         case OBJ_SEARCH_OBJECT:
453                 right_key = tps_right->name;
454                 /* Fall through */
455         case OBJ_SEARCH_KEY:
456                 cmp = strcasecmp(tps_left->name, right_key);
457                 break;
458         case OBJ_SEARCH_PARTIAL_KEY:
459                 cmp = strncasecmp(tps_left->name, right_key, strlen(right_key));
460                 break;
461         }
462         return cmp;
463 }
464
465 static char *cli_tps_report(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
466 {
467         char name[256];
468         int tcount;
469         unsigned long qsize;
470         unsigned long maxqsize;
471         unsigned long processed;
472         struct ao2_container *sorted_tps;
473         struct ast_taskprocessor *tps;
474         struct ao2_iterator iter;
475 #define FMT_HEADERS             "%-45s %10s %10s %10s %10s %10s\n"
476 #define FMT_FIELDS              "%-45s %10lu %10lu %10lu %10lu %10lu\n"
477
478         switch (cmd) {
479         case CLI_INIT:
480                 e->command = "core show taskprocessors";
481                 e->usage =
482                         "Usage: core show taskprocessors\n"
483                         "       Shows a list of instantiated task processors and their statistics\n";
484                 return NULL;
485         case CLI_GENERATE:
486                 return NULL;
487         }
488
489         if (a->argc != e->args) {
490                 return CLI_SHOWUSAGE;
491         }
492
493         sorted_tps = ao2_container_alloc_rbtree(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, tps_sort_cb,
494                 NULL);
495         if (!sorted_tps
496                 || ao2_container_dup(sorted_tps, tps_singletons, 0)) {
497                 ao2_cleanup(sorted_tps);
498                 return CLI_FAILURE;
499         }
500
501         ast_cli(a->fd, "\n" FMT_HEADERS, "Processor", "Processed", "In Queue", "Max Depth", "Low water", "High water");
502         tcount = 0;
503         iter = ao2_iterator_init(sorted_tps, AO2_ITERATOR_UNLINK);
504         while ((tps = ao2_iterator_next(&iter))) {
505                 ast_copy_string(name, tps->name, sizeof(name));
506                 qsize = tps->tps_queue_size;
507                 if (tps->stats) {
508                         maxqsize = tps->stats->max_qsize;
509                         processed = tps->stats->_tasks_processed_count;
510                 } else {
511                         maxqsize = 0;
512                         processed = 0;
513                 }
514                 ast_cli(a->fd, FMT_FIELDS, name, processed, qsize, maxqsize,
515                         tps->tps_queue_low, tps->tps_queue_high);
516                 ast_taskprocessor_unreference(tps);
517                 ++tcount;
518         }
519         ao2_iterator_destroy(&iter);
520         ast_cli(a->fd, "\n%d taskprocessors\n\n", tcount);
521         ao2_ref(sorted_tps, -1);
522         return CLI_SUCCESS;
523 }
524
525 /* hash callback for astobj2 */
526 static int tps_hash_cb(const void *obj, const int flags)
527 {
528         const struct ast_taskprocessor *tps = obj;
529         const char *name = flags & OBJ_KEY ? obj : tps->name;
530
531         return ast_str_case_hash(name);
532 }
533
534 /* compare callback for astobj2 */
535 static int tps_cmp_cb(void *obj, void *arg, int flags)
536 {
537         struct ast_taskprocessor *lhs = obj, *rhs = arg;
538         const char *rhsname = flags & OBJ_KEY ? arg : rhs->name;
539
540         return !strcasecmp(lhs->name, rhsname) ? CMP_MATCH | CMP_STOP : 0;
541 }
542
543 /*! Count of the number of taskprocessors in high water alert. */
544 static unsigned int tps_alert_count;
545
546 /*! Access protection for tps_alert_count */
547 AST_RWLOCK_DEFINE_STATIC(tps_alert_lock);
548
549 /*!
550  * \internal
551  * \brief Add a delta to tps_alert_count with protection.
552  * \since 13.10.0
553  *
554  * \param tps Taskprocessor updating queue water mark alert trigger.
555  * \param delta The amount to add to tps_alert_count.
556  *
557  * \return Nothing
558  */
559 static void tps_alert_add(struct ast_taskprocessor *tps, int delta)
560 {
561         unsigned int old;
562
563         ast_rwlock_wrlock(&tps_alert_lock);
564         old = tps_alert_count;
565         tps_alert_count += delta;
566         if (DEBUG_ATLEAST(3)
567                 /* and tps_alert_count becomes zero or non-zero */
568                 && !old != !tps_alert_count) {
569                 ast_log(LOG_DEBUG, "Taskprocessor '%s' %s the high water alert.\n",
570                         tps->name, tps_alert_count ? "triggered" : "cleared");
571         }
572         ast_rwlock_unlock(&tps_alert_lock);
573 }
574
575 unsigned int ast_taskprocessor_alert_get(void)
576 {
577         unsigned int count;
578
579         ast_rwlock_rdlock(&tps_alert_lock);
580         count = tps_alert_count;
581         ast_rwlock_unlock(&tps_alert_lock);
582
583         return count;
584 }
585
586 int ast_taskprocessor_alert_set_levels(struct ast_taskprocessor *tps, long low_water, long high_water)
587 {
588         if (!tps || high_water < 0 || high_water < low_water) {
589                 return -1;
590         }
591
592         if (low_water < 0) {
593                 /* Set low water level to 90% of high water level */
594                 low_water = (high_water * 9) / 10;
595         }
596
597         ao2_lock(tps);
598
599         tps->tps_queue_low = low_water;
600         tps->tps_queue_high = high_water;
601
602         if (tps->high_water_alert) {
603                 if (!tps->tps_queue_size || tps->tps_queue_size < low_water) {
604                         /* Update water mark alert immediately */
605                         tps->high_water_alert = 0;
606                         tps_alert_add(tps, -1);
607                 }
608         } else {
609                 if (high_water < tps->tps_queue_size) {
610                         /* Update water mark alert immediately */
611                         tps->high_water_alert = 1;
612                         tps_alert_add(tps, +1);
613                 }
614         }
615
616         ao2_unlock(tps);
617
618         return 0;
619 }
620
621 /* destroy the taskprocessor */
622 static void tps_taskprocessor_dtor(void *tps)
623 {
624         struct ast_taskprocessor *t = tps;
625         struct tps_task *task;
626
627         while ((task = AST_LIST_REMOVE_HEAD(&t->tps_queue, list))) {
628                 tps_task_free(task);
629         }
630         t->tps_queue_size = 0;
631
632         if (t->high_water_alert) {
633                 t->high_water_alert = 0;
634                 tps_alert_add(t, -1);
635         }
636
637         ast_free(t->stats);
638         t->stats = NULL;
639         ast_free((char *) t->name);
640         t->name = NULL;
641         ao2_cleanup(t->listener);
642         t->listener = NULL;
643 }
644
645 /* pop the front task and return it */
646 static struct tps_task *tps_taskprocessor_pop(struct ast_taskprocessor *tps)
647 {
648         struct tps_task *task;
649
650         if ((task = AST_LIST_REMOVE_HEAD(&tps->tps_queue, list))) {
651                 --tps->tps_queue_size;
652                 if (tps->high_water_alert && tps->tps_queue_size <= tps->tps_queue_low) {
653                         tps->high_water_alert = 0;
654                         tps_alert_add(tps, -1);
655                 }
656         }
657         return task;
658 }
659
660 long ast_taskprocessor_size(struct ast_taskprocessor *tps)
661 {
662         return (tps) ? tps->tps_queue_size : -1;
663 }
664
665 /* taskprocessor name accessor */
666 const char *ast_taskprocessor_name(struct ast_taskprocessor *tps)
667 {
668         if (!tps) {
669                 ast_log(LOG_ERROR, "no taskprocessor specified!\n");
670                 return NULL;
671         }
672         return tps->name;
673 }
674
675 static void listener_shutdown(struct ast_taskprocessor_listener *listener)
676 {
677         listener->callbacks->shutdown(listener);
678         ao2_ref(listener->tps, -1);
679 }
680
681 static void taskprocessor_listener_dtor(void *obj)
682 {
683         struct ast_taskprocessor_listener *listener = obj;
684
685         if (listener->callbacks->dtor) {
686                 listener->callbacks->dtor(listener);
687         }
688 }
689
690 struct ast_taskprocessor_listener *ast_taskprocessor_listener_alloc(const struct ast_taskprocessor_listener_callbacks *callbacks, void *user_data)
691 {
692         struct ast_taskprocessor_listener *listener;
693
694         listener = ao2_alloc(sizeof(*listener), taskprocessor_listener_dtor);
695         if (!listener) {
696                 return NULL;
697         }
698         listener->callbacks = callbacks;
699         listener->user_data = user_data;
700
701         return listener;
702 }
703
704 struct ast_taskprocessor *ast_taskprocessor_listener_get_tps(const struct ast_taskprocessor_listener *listener)
705 {
706         ao2_ref(listener->tps, +1);
707         return listener->tps;
708 }
709
710 void *ast_taskprocessor_listener_get_user_data(const struct ast_taskprocessor_listener *listener)
711 {
712         return listener->user_data;
713 }
714
715 static void *default_listener_pvt_alloc(void)
716 {
717         struct default_taskprocessor_listener_pvt *pvt;
718
719         pvt = ast_calloc(1, sizeof(*pvt));
720         if (!pvt) {
721                 return NULL;
722         }
723         pvt->poll_thread = AST_PTHREADT_NULL;
724         if (ast_sem_init(&pvt->sem, 0, 0) != 0) {
725                 ast_log(LOG_ERROR, "ast_sem_init(): %s\n", strerror(errno));
726                 ast_free(pvt);
727                 return NULL;
728         }
729         return pvt;
730 }
731
732 static struct ast_taskprocessor *__allocate_taskprocessor(const char *name, struct ast_taskprocessor_listener *listener)
733 {
734         struct ast_taskprocessor *p;
735
736         p = ao2_alloc(sizeof(*p), tps_taskprocessor_dtor);
737         if (!p) {
738                 ast_log(LOG_WARNING, "failed to create taskprocessor '%s'\n", name);
739                 return NULL;
740         }
741
742         /* Set default congestion water level alert triggers. */
743         p->tps_queue_low = (AST_TASKPROCESSOR_HIGH_WATER_LEVEL * 9) / 10;
744         p->tps_queue_high = AST_TASKPROCESSOR_HIGH_WATER_LEVEL;
745
746         p->stats = ast_calloc(1, sizeof(*p->stats));
747         p->name = ast_strdup(name);
748         if (!p->stats || !p->name) {
749                 ao2_ref(p, -1);
750                 return NULL;
751         }
752
753         ao2_ref(listener, +1);
754         p->listener = listener;
755
756         p->thread = AST_PTHREADT_NULL;
757
758         ao2_ref(p, +1);
759         listener->tps = p;
760
761         if (!(ao2_link(tps_singletons, p))) {
762                 ast_log(LOG_ERROR, "Failed to add taskprocessor '%s' to container\n", p->name);
763                 listener->tps = NULL;
764                 ao2_ref(p, -2);
765                 return NULL;
766         }
767
768         if (p->listener->callbacks->start(p->listener)) {
769                 ast_log(LOG_ERROR, "Unable to start taskprocessor listener for taskprocessor %s\n",
770                         p->name);
771                 ast_taskprocessor_unreference(p);
772                 return NULL;
773         }
774
775         return p;
776 }
777
778 /* Provide a reference to a taskprocessor.  Create the taskprocessor if necessary, but don't
779  * create the taskprocessor if we were told via ast_tps_options to return a reference only
780  * if it already exists */
781 struct ast_taskprocessor *ast_taskprocessor_get(const char *name, enum ast_tps_options create)
782 {
783         struct ast_taskprocessor *p;
784         struct ast_taskprocessor_listener *listener;
785         struct default_taskprocessor_listener_pvt *pvt;
786
787         if (ast_strlen_zero(name)) {
788                 ast_log(LOG_ERROR, "requesting a nameless taskprocessor!!!\n");
789                 return NULL;
790         }
791         p = ao2_find(tps_singletons, name, OBJ_KEY);
792         if (p) {
793                 return p;
794         }
795         if (create & TPS_REF_IF_EXISTS) {
796                 /* calling function does not want a new taskprocessor to be created if it doesn't already exist */
797                 return NULL;
798         }
799         /* Create a new taskprocessor. Start by creating a default listener */
800         pvt = default_listener_pvt_alloc();
801         if (!pvt) {
802                 return NULL;
803         }
804         listener = ast_taskprocessor_listener_alloc(&default_listener_callbacks, pvt);
805         if (!listener) {
806                 default_listener_pvt_destroy(pvt);
807                 return NULL;
808         }
809
810         p = __allocate_taskprocessor(name, listener);
811
812         ao2_ref(listener, -1);
813         return p;
814 }
815
816 struct ast_taskprocessor *ast_taskprocessor_create_with_listener(const char *name, struct ast_taskprocessor_listener *listener)
817 {
818         struct ast_taskprocessor *p = ao2_find(tps_singletons, name, OBJ_KEY);
819
820         if (p) {
821                 ast_taskprocessor_unreference(p);
822                 return NULL;
823         }
824         return __allocate_taskprocessor(name, listener);
825 }
826
827 void ast_taskprocessor_set_local(struct ast_taskprocessor *tps,
828         void *local_data)
829 {
830         SCOPED_AO2LOCK(lock, tps);
831         tps->local_data = local_data;
832 }
833
834 /* decrement the taskprocessor reference count and unlink from the container if necessary */
835 void *ast_taskprocessor_unreference(struct ast_taskprocessor *tps)
836 {
837         if (!tps) {
838                 return NULL;
839         }
840
841         /* To prevent another thread from finding and getting a reference to this
842          * taskprocessor we hold the singletons lock. If we didn't do this then
843          * they may acquire it and find that the listener has been shut down.
844          */
845         ao2_lock(tps_singletons);
846
847         if (ao2_ref(tps, -1) > 3) {
848                 ao2_unlock(tps_singletons);
849                 return NULL;
850         }
851
852         /* If we're down to 3 references, then those must be:
853          * 1. The reference we just got rid of
854          * 2. The container
855          * 3. The listener
856          */
857         ao2_unlink_flags(tps_singletons, tps, OBJ_NOLOCK);
858         ao2_unlock(tps_singletons);
859
860         listener_shutdown(tps->listener);
861         return NULL;
862 }
863
864 /* push the task into the taskprocessor queue */
865 static int taskprocessor_push(struct ast_taskprocessor *tps, struct tps_task *t)
866 {
867         int previous_size;
868         int was_empty;
869
870         if (!tps) {
871                 ast_log(LOG_ERROR, "tps is NULL!\n");
872                 return -1;
873         }
874
875         if (!t) {
876                 ast_log(LOG_ERROR, "t is NULL!\n");
877                 return -1;
878         }
879
880         ao2_lock(tps);
881         AST_LIST_INSERT_TAIL(&tps->tps_queue, t, list);
882         previous_size = tps->tps_queue_size++;
883
884         if (tps->tps_queue_high <= tps->tps_queue_size) {
885                 if (!tps->high_water_alert) {
886                         ast_log(LOG_WARNING, "The '%s' task processor queue reached %ld scheduled tasks%s.\n",
887                                 tps->name, tps->tps_queue_size, tps->high_water_warned ? " again" : "");
888                         tps->high_water_warned = 1;
889                         tps->high_water_alert = 1;
890                         tps_alert_add(tps, +1);
891                 }
892         }
893
894         /* The currently executing task counts as still in queue */
895         was_empty = tps->executing ? 0 : previous_size == 0;
896         ao2_unlock(tps);
897         tps->listener->callbacks->task_pushed(tps->listener, was_empty);
898         return 0;
899 }
900
901 int ast_taskprocessor_push(struct ast_taskprocessor *tps, int (*task_exe)(void *datap), void *datap)
902 {
903         return taskprocessor_push(tps, tps_task_alloc(task_exe, datap));
904 }
905
906 int ast_taskprocessor_push_local(struct ast_taskprocessor *tps, int (*task_exe)(struct ast_taskprocessor_local *datap), void *datap)
907 {
908         return taskprocessor_push(tps, tps_task_alloc_local(task_exe, datap));
909 }
910
911 int ast_taskprocessor_suspend(struct ast_taskprocessor *tps)
912 {
913         if (tps) {
914                 ao2_lock(tps);
915                 tps->suspended = 1;
916                 ao2_unlock(tps);
917                 return 0;
918         }
919         return -1;
920 }
921
922 int ast_taskprocessor_unsuspend(struct ast_taskprocessor *tps)
923 {
924         if (tps) {
925                 ao2_lock(tps);
926                 tps->suspended = 0;
927                 ao2_unlock(tps);
928                 return 0;
929         }
930         return -1;
931 }
932
933 int ast_taskprocessor_is_suspended(struct ast_taskprocessor *tps)
934 {
935         return tps ? tps->suspended : -1;
936 }
937
938 int ast_taskprocessor_execute(struct ast_taskprocessor *tps)
939 {
940         struct ast_taskprocessor_local local;
941         struct tps_task *t;
942         long size;
943
944         ao2_lock(tps);
945         t = tps_taskprocessor_pop(tps);
946         if (!t) {
947                 ao2_unlock(tps);
948                 return 0;
949         }
950
951         tps->thread = pthread_self();
952         tps->executing = 1;
953
954         if (t->wants_local) {
955                 local.local_data = tps->local_data;
956                 local.data = t->datap;
957         }
958         ao2_unlock(tps);
959
960         if (t->wants_local) {
961                 t->callback.execute_local(&local);
962         } else {
963                 t->callback.execute(t->datap);
964         }
965         tps_task_free(t);
966
967         ao2_lock(tps);
968         tps->thread = AST_PTHREADT_NULL;
969         /* We need to check size in the same critical section where we reset the
970          * executing bit. Avoids a race condition where a task is pushed right
971          * after we pop an empty stack.
972          */
973         tps->executing = 0;
974         size = ast_taskprocessor_size(tps);
975
976         /* Update the stats */
977         if (tps->stats) {
978                 ++tps->stats->_tasks_processed_count;
979
980                 /* Include the task we just executed as part of the queue size. */
981                 if (size >= tps->stats->max_qsize) {
982                         tps->stats->max_qsize = size + 1;
983                 }
984         }
985         ao2_unlock(tps);
986
987         /* If we executed a task, check for the transition to empty */
988         if (size == 0 && tps->listener->callbacks->emptied) {
989                 tps->listener->callbacks->emptied(tps->listener);
990         }
991         return size > 0;
992 }
993
994 int ast_taskprocessor_is_task(struct ast_taskprocessor *tps)
995 {
996         int is_task;
997
998         ao2_lock(tps);
999         is_task = pthread_equal(tps->thread, pthread_self());
1000         ao2_unlock(tps);
1001         return is_task;
1002 }
1003
1004 unsigned int ast_taskprocessor_seq_num(void)
1005 {
1006         static int seq_num;
1007
1008         return (unsigned int) ast_atomic_fetchadd_int(&seq_num, +1);
1009 }
1010
1011 void ast_taskprocessor_build_name(char *buf, unsigned int size, const char *format, ...)
1012 {
1013         va_list ap;
1014         int user_size;
1015 #define SEQ_STR_SIZE (1 + 8 + 1)        /* Dash plus 8 hex digits plus null terminator */
1016
1017         ast_assert(buf != NULL);
1018         ast_assert(SEQ_STR_SIZE <= size);
1019
1020         va_start(ap, format);
1021         user_size = vsnprintf(buf, size - (SEQ_STR_SIZE - 1), format, ap);
1022         va_end(ap);
1023         if (user_size < 0) {
1024                 /*
1025                  * Wow!  We got an output error to a memory buffer.
1026                  * Assume no user part of name written.
1027                  */
1028                 user_size = 0;
1029         } else if (size < user_size + SEQ_STR_SIZE) {
1030                 /* Truncate user part of name to make sequence number fit. */
1031                 user_size = size - SEQ_STR_SIZE;
1032         }
1033
1034         /* Append sequence number to end of user name. */
1035         snprintf(buf + user_size, SEQ_STR_SIZE, "-%08x", ast_taskprocessor_seq_num());
1036 }