[svn-commits] juggie: branch group/NoLossCDR-Redux r104104 - /team/group/NoLossCDR-Redux/main/
SVN commits to the Digium repositories
svn-commits at lists.digium.com
Mon Feb 25 17:19:53 CST 2008
Author: juggie
Date: Mon Feb 25 17:19:52 2008
New Revision: 104104
URL: http://svn.digium.com/view/asterisk?view=rev&rev=104104
Log:
clean up batch code, and other misc stuff
Modified:
team/group/NoLossCDR-Redux/main/cdr.c
Modified: team/group/NoLossCDR-Redux/main/cdr.c
URL: http://svn.digium.com/view/asterisk/team/group/NoLossCDR-Redux/main/cdr.c?view=diff&rev=104104&r1=104103&r2=104104
==============================================================================
--- team/group/NoLossCDR-Redux/main/cdr.c (original)
+++ team/group/NoLossCDR-Redux/main/cdr.c Mon Feb 25 17:19:52 2008
@@ -73,39 +73,15 @@
static AST_RWLIST_HEAD_STATIC(backends, ast_cdr_backend);
-struct ast_cdr_batch_item {
- struct ast_cdr *cdr;
- struct ast_cdr_batch_item *next;
-};
-
-static struct ast_cdr_batch {
- int size;
- struct ast_cdr_batch_item *head;
- struct ast_cdr_batch_item *tail;
-} *batch = NULL;
-
-static struct sched_context *sched;
-static int cdr_sched = -1;
-static pthread_t cdr_thread = AST_PTHREADT_NULL;
-
-#define BATCH_SIZE_DEFAULT 100
-#define BATCH_TIME_DEFAULT 300
-#define BATCH_SCHEDULER_ONLY_DEFAULT 0
-#define BATCH_SAFE_SHUTDOWN_DEFAULT 1
+#define SAFE_SHUTDOWN_DEFAULT 1
+#define MAX_QUEUE_SIZE_DEFAULT 1000
+#define RETRY_DELAY_DEFAULT 30
static int enabled; /*! Is the CDR subsystem enabled ? */
static int unanswered;
-static int batchmode;
-static int batchsize;
-static int batchtime;
-static int batchscheduleronly;
-static int batchsafeshutdown;
-
-AST_MUTEX_DEFINE_STATIC(cdr_batch_lock);
-
-/* these are used to wake up the CDR thread when there's work to do */
-AST_MUTEX_DEFINE_STATIC(cdr_pending_lock);
-static ast_cond_t cdr_pending_cond;
+static int safeshutdown;
+static int queuesize;
+static int retrydelay;
int check_cdr_enabled()
{
@@ -1127,99 +1103,8 @@
return ret;
}
-/*! \note Don't call without cdr_batch_lock */
-static void reset_batch(void)
-{
- batch->size = 0;
- batch->head = NULL;
- batch->tail = NULL;
-}
-
-/*! \note Don't call without cdr_batch_lock */
-static int init_batch(void)
-{
- /* This is the single meta-batch used to keep track of all CDRs during the entire life of the program */
- if (!(batch = ast_malloc(sizeof(*batch))))
- return -1;
-
- reset_batch();
-
- return 0;
-}
-
-static void *do_batch_backend_process(void *data)
-{
- struct ast_cdr_batch_item *processeditem;
- struct ast_cdr_batch_item *batchitem = data;
-
- /* Push each CDR into storage mechanism(s) and free all the memory */
- while (batchitem) {
- post_cdr(batchitem->cdr);
- ast_cdr_free(batchitem->cdr);
- processeditem = batchitem;
- batchitem = batchitem->next;
- ast_free(processeditem);
- }
-
- return NULL;
-}
-
-void ast_cdr_submit_batch(int shutdown)
-{
- struct ast_cdr_batch_item *oldbatchitems = NULL;
- pthread_t batch_post_thread = AST_PTHREADT_NULL;
-
- /* if there's no batch, or no CDRs in the batch, then there's nothing to do */
- if (!batch || !batch->head)
- return;
-
- /* move the old CDRs aside, and prepare a new CDR batch */
- ast_mutex_lock(&cdr_batch_lock);
- oldbatchitems = batch->head;
- reset_batch();
- ast_mutex_unlock(&cdr_batch_lock);
-
- /* if configured, spawn a new thread to post these CDRs,
- also try to save as much as possible if we are shutting down safely */
- if (batchscheduleronly || shutdown) {
- ast_debug(1, "CDR single-threaded batch processing begins now\n");
- do_batch_backend_process(oldbatchitems);
- } else {
- if (ast_pthread_create_detached_background(&batch_post_thread, NULL, do_batch_backend_process, oldbatchitems)) {
- ast_log(LOG_WARNING, "CDR processing thread could not detach, now trying in this thread\n");
- do_batch_backend_process(oldbatchitems);
- } else {
- ast_debug(1, "CDR multi-threaded batch processing begins now\n");
- }
- }
-}
-
-static int submit_scheduled_batch(const void *data)
-{
- ast_cdr_submit_batch(0);
- /* manually reschedule from this point in time */
- cdr_sched = ast_sched_add(sched, batchtime * 1000, submit_scheduled_batch, NULL);
- /* returning zero so the scheduler does not automatically reschedule */
- return 0;
-}
-
-static void submit_unscheduled_batch(void)
-{
- /* this is okay since we are not being called from within the scheduler */
- AST_SCHED_DEL(sched, cdr_sched);
- /* schedule the submission to occur ASAP (1 ms) */
- cdr_sched = ast_sched_add(sched, 1, submit_scheduled_batch, NULL);
- /* signal the do_cdr thread to wakeup early and do some work (that lazy thread ;) */
- ast_mutex_lock(&cdr_pending_lock);
- ast_cond_signal(&cdr_pending_cond);
- ast_mutex_unlock(&cdr_pending_lock);
-}
-
void ast_cdr_detach(struct ast_cdr *cdr)
{
- struct ast_cdr_batch_item *newtail;
- int curr;
-
if (!cdr)
return;
@@ -1231,65 +1116,13 @@
return;
}
- /* post stuff immediately if we are not in batch mode, this is legacy behaviour */
- if (!batchmode) {
- post_cdr(cdr);
- ast_cdr_free(cdr);
- return;
- }
-
- /* otherwise, each CDR gets put into a batch list (at the end) */
- ast_debug(1, "CDR detaching from this thread\n");
-
- /* we'll need a new tail for every CDR */
- if (!(newtail = ast_calloc(1, sizeof(*newtail)))) {
- post_cdr(cdr);
- ast_cdr_free(cdr);
- return;
- }
-
- /* don't traverse a whole list (just keep track of the tail) */
- ast_mutex_lock(&cdr_batch_lock);
- if (!batch)
- init_batch();
- if (!batch->head) {
- /* new batch is empty, so point the head at the new tail */
- batch->head = newtail;
- } else {
- /* already got a batch with something in it, so just append a new tail */
- batch->tail->next = newtail;
- }
- newtail->cdr = cdr;
- batch->tail = newtail;
- curr = batch->size++;
- ast_mutex_unlock(&cdr_batch_lock);
-
- /* if we have enough stuff to post, then do it */
- if (curr >= (batchsize - 1))
- submit_unscheduled_batch();
+ post_cdr(cdr);
}
static void *do_cdr(void *data)
{
- struct timespec timeout;
- int schedms;
- int numevents = 0;
-
for (;;) {
- struct timeval now;
- schedms = ast_sched_wait(sched);
- /* this shouldn't happen, but provide a 1 second default just in case */
- if (schedms <= 0)
- schedms = 1000;
- now = ast_tvadd(ast_tvnow(), ast_samp2tv(schedms, 1000));
- timeout.tv_sec = now.tv_sec;
- timeout.tv_nsec = now.tv_usec * 1000;
- /* prevent stuff from clobbering cdr_pending_cond, then wait on signals sent to it until the timeout expires */
- ast_mutex_lock(&cdr_pending_lock);
- ast_cond_timedwait(&cdr_pending_cond, &cdr_pending_lock, &timeout);
- numevents = ast_sched_runq(sched);
- ast_mutex_unlock(&cdr_pending_lock);
- ast_debug(2, "Processed %d scheduled CDR batches from the run queue\n", numevents);
+ sleep(1000);
}
return NULL;
@@ -1299,8 +1132,6 @@
{
struct ast_cdr_backend *backend = NULL;
struct ast_cdr_backend_sink *sink = NULL;
- int cnt = 0;
- long nextbatchtime = 0;
switch (cmd) {
case CLI_INIT:
@@ -1317,21 +1148,8 @@
return CLI_SHOWUSAGE;
ast_cli(a->fd, "CDR logging: %s\n", enabled ? "enabled" : "disabled");
- ast_cli(a->fd, "CDR mode: %s\n", batchmode ? "batch" : "simple");
if (enabled) {
ast_cli(a->fd, "CDR output unanswered calls: %s\n", unanswered ? "yes" : "no");
- if (batchmode) {
- if (batch)
- cnt = batch->size;
- if (cdr_sched > -1)
- nextbatchtime = ast_sched_when(sched, cdr_sched);
- ast_cli(a->fd, "CDR safe shut down: %s\n", batchsafeshutdown ? "enabled" : "disabled");
- ast_cli(a->fd, "CDR batch threading model: %s\n", batchscheduleronly ? "scheduler only" : "scheduler plus separate threads");
- ast_cli(a->fd, "CDR current batch size: %d record%s\n", cnt, ESS(cnt));
- ast_cli(a->fd, "CDR maximum batch size: %d record%s\n", batchsize, ESS(batchsize));
- ast_cli(a->fd, "CDR maximum batch time: %d second%s\n", batchtime, ESS(batchtime));
- ast_cli(a->fd, "CDR next scheduled batch processing time: %ld second%s\n", nextbatchtime, ESS(nextbatchtime));
- }
ast_cli(a->fd, "\n");
ast_cli(a->fd, "Registered Backends\n");
ast_cli(a->fd, "-------------------\n");
@@ -1360,28 +1178,6 @@
return res;
}
-static char *handle_cli_submit(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
-{
- switch (cmd) {
- case CLI_INIT:
- e->command = "cdr submit";
- e->usage =
- "Usage: cdr submit\n"
- " Posts all pending batched CDR data to the configured CDR backend engine modules.\n";
- return NULL;
- case CLI_GENERATE:
- return NULL;
- }
- if (a->argc > 2)
- return CLI_SHOWUSAGE;
-
- submit_unscheduled_batch();
- ast_cli(a->fd, "Submitted CDRs to backend engines for processing. This may take a while.\n");
-
- return CLI_SUCCESS;
-}
-
-static struct ast_cli_entry cli_submit = AST_CLI_DEFINE(handle_cli_submit, "Posts all pending batched CDR data");
static struct ast_cli_entry cli_status_deprecated = AST_CLI_DEFINE(handle_cli_status_deprecated, "Display the CDR status");
static struct ast_cli_entry cli_status = AST_CLI_DEFINE(handle_cli_status, "Display the CDR status", .deprecate_cmd = &cli_status_deprecated);
@@ -1390,35 +1186,16 @@
struct ast_config *config;
const char *enabled_value;
const char *unanswered_value;
- const char *batched_value;
- const char *scheduleronly_value;
- const char *batchsafeshutdown_value;
- const char *size_value;
- const char *time_value;
const char *end_before_h_value;
- int cfg_size;
- int cfg_time;
int was_enabled;
- int was_batchmode;
int res=0;
struct ast_flags config_flags = { reload ? CONFIG_FLAG_FILEUNCHANGED : 0 };
if ((config = ast_config_load("cdr.conf", config_flags)) == CONFIG_STATUS_FILEUNCHANGED)
return 0;
- ast_mutex_lock(&cdr_batch_lock);
-
- batchsize = BATCH_SIZE_DEFAULT;
- batchtime = BATCH_TIME_DEFAULT;
- batchscheduleronly = BATCH_SCHEDULER_ONLY_DEFAULT;
- batchsafeshutdown = BATCH_SAFE_SHUTDOWN_DEFAULT;
was_enabled = enabled;
- was_batchmode = batchmode;
enabled = 1;
- batchmode = 0;
-
- /* don't run the next scheduled CDR posting while reloading */
- AST_SCHED_DEL(sched, cdr_sched);
if (config) {
if ((enabled_value = ast_variable_retrieve(config, "general", "enable"))) {
@@ -1427,78 +1204,16 @@
if ((unanswered_value = ast_variable_retrieve(config, "general", "unanswered"))) {
unanswered = ast_true(unanswered_value);
}
- if ((batched_value = ast_variable_retrieve(config, "general", "batch"))) {
- batchmode = ast_true(batched_value);
- }
- if ((scheduleronly_value = ast_variable_retrieve(config, "general", "scheduleronly"))) {
- batchscheduleronly = ast_true(scheduleronly_value);
- }
- if ((batchsafeshutdown_value = ast_variable_retrieve(config, "general", "safeshutdown"))) {
- batchsafeshutdown = ast_true(batchsafeshutdown_value);
- }
- if ((size_value = ast_variable_retrieve(config, "general", "size"))) {
- if (sscanf(size_value, "%d", &cfg_size) < 1)
- ast_log(LOG_WARNING, "Unable to convert '%s' to a numeric value.\n", size_value);
- else if (size_value < 0)
- ast_log(LOG_WARNING, "Invalid maximum batch size '%d' specified, using default\n", cfg_size);
- else
- batchsize = cfg_size;
- }
- if ((time_value = ast_variable_retrieve(config, "general", "time"))) {
- if (sscanf(time_value, "%d", &cfg_time) < 1)
- ast_log(LOG_WARNING, "Unable to convert '%s' to a numeric value.\n", time_value);
- else if (time_value < 0)
- ast_log(LOG_WARNING, "Invalid maximum batch time '%d' specified, using default\n", cfg_time);
- else
- batchtime = cfg_time;
- }
if ((end_before_h_value = ast_variable_retrieve(config, "general", "endbeforehexten")))
ast_set2_flag(&ast_options, ast_true(end_before_h_value), AST_OPT_FLAG_END_CDR_BEFORE_H_EXTEN);
}
- if (enabled && !batchmode) {
- ast_log(LOG_NOTICE, "CDR simple logging enabled.\n");
- } else if (enabled && batchmode) {
- cdr_sched = ast_sched_add(sched, batchtime * 1000, submit_scheduled_batch, NULL);
- ast_log(LOG_NOTICE, "CDR batch mode logging enabled, first of either size %d or time %d seconds.\n", batchsize, batchtime);
+ if (enabled) {
+ ast_log(LOG_NOTICE, "CDR logging enabled.\n");
} else {
ast_log(LOG_NOTICE, "CDR logging disabled, data will be lost.\n");
}
- /* if this reload enabled the CDR batch mode, create the background thread
- if it does not exist */
- if (enabled && batchmode && (!was_enabled || !was_batchmode) && (cdr_thread == AST_PTHREADT_NULL)) {
- ast_cond_init(&cdr_pending_cond, NULL);
- if (ast_pthread_create_background(&cdr_thread, NULL, do_cdr, NULL) < 0) {
- ast_log(LOG_ERROR, "Unable to start CDR thread.\n");
- AST_SCHED_DEL(sched, cdr_sched);
- } else {
- ast_cli_register(&cli_submit);
- ast_register_atexit(ast_cdr_engine_term);
- res = 0;
- }
- /* if this reload disabled the CDR and/or batch mode and there is a background thread,
- kill it */
- } else if (((!enabled && was_enabled) || (!batchmode && was_batchmode)) && (cdr_thread != AST_PTHREADT_NULL)) {
- /* wake up the thread so it will exit */
- pthread_cancel(cdr_thread);
- pthread_kill(cdr_thread, SIGURG);
- pthread_join(cdr_thread, NULL);
- cdr_thread = AST_PTHREADT_NULL;
- ast_cond_destroy(&cdr_pending_cond);
- ast_cli_unregister(&cli_submit);
- ast_unregister_atexit(ast_cdr_engine_term);
- res = 0;
- /* if leaving batch mode, then post the CDRs in the batch,
- and don't reschedule, since we are stopping CDR logging */
- if (!batchmode && was_batchmode) {
- ast_cdr_engine_term();
- }
- } else {
- res = 0;
- }
-
- ast_mutex_unlock(&cdr_batch_lock);
ast_config_destroy(config);
manager_event(EVENT_FLAG_SYSTEM, "Reload", "Module: CDR\r\nMessage: CDR subsystem reload requested\r\n");
@@ -1507,31 +1222,15 @@
int ast_cdr_engine_init(void)
{
- int res;
-
- sched = sched_context_create();
- if (!sched) {
- ast_log(LOG_ERROR, "Unable to create schedule context.\n");
- return -1;
- }
-
ast_cli_register(&cli_status);
-
- res = do_reload(0);
- if (res) {
- ast_mutex_lock(&cdr_batch_lock);
- res = init_batch();
- ast_mutex_unlock(&cdr_batch_lock);
- }
-
- return res;
+ return do_reload(0);
}
/* \note This actually gets called a couple of times at shutdown. Once, before we start
hanging up channels, and then again, after the channel hangup timeout expires */
void ast_cdr_engine_term(void)
{
- ast_cdr_submit_batch(batchsafeshutdown);
+/* safeshutdown code goes here */
}
int ast_cdr_engine_reload(void)
More information about the svn-commits
mailing list