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