source: trunk/src/arm4_daemon.cpp @ 704

/**********************************************************************
 * Copyright (c) 2005-2008 David Carter <dcarter@arm4.org> and others.
 * All rights reserved.   This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * Contributors:
 * David Carter - Initial API and implementation
 **********************************************************************/

#include "config.h"

#include <sys/types.h>
#include <sys/time.h>
#include <errno.h>

#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <math.h>
#include <signal.h>
#include <pthread.h>
#include <syslog.h>
#include <time.h>

#ifdef HAVE_GETOPT_H
#include <getopt.h>
#endif

#include <db.h>
#include "arm4.h"
#include "Arm4db.h"
#include "Logger.h"
#include "Arm4dbConfig.h"
#include "Arm4dbCommon.h"
#include "Arm4dbDaemonSharedMemory.h"
#include "arm4agent.h"

/* Prototypes */
static void on_close (void);
static void process_queues (void);
static void usage (void);

/* Flag set by `--verbose'. */
static int g_verbose = 0;
static int g_debug = 0;
static int g_stats = 0;
static int g_off = 0;

static void
arm_sleep (int seconds)
{
        /* Implements the sleep function using nanosleep */
        struct timespec required;

        required.tv_sec = seconds;
        required.tv_nsec = 0;

        nanosleep (&required, NULL);
}

static void
backup (void)
{
    if (g_verbose)
        printf ("Starting backup...\n");

        try
        {
                Arm4db::getArm4db ()->backup ();
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "backup");
        }

    if (g_verbose)
        printf ("... complete.\n");
}

static void
checkpoint (void)
{
    if (g_verbose)
        printf ("Starting checkpoint...\n");

        try
        {
                Arm4db::getArm4db ()->checkpoint ();
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "backup");
        }

    if (g_verbose)
        printf ("... complete.\n");
}

static void
termination_handler (int signum)
{
        /*on_close ();*/

        /* Should also message the threads that we're aborting */
        exit (0);
}

static void
on_close (void)
{
    /* Turn collection off so that programs can continue */
    Arm4dbDaemonSharedMemory::setNullCollector (ARM_TRUE);

    if (g_verbose)
                printf ("Closing databases...\n");

        try
        {
                Arm4dbDaemonSharedMemory::sampleFlush ();
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "on_close");
        }

        try
        {
                Arm4db::getArm4db ()->checkpoint ();
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "on_close");
        }

        try
        {
                Arm4db::getArm4db ()->closeDatabases ();
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "on_close");
        }

        if (g_verbose)
                printf ("...done\n");
}

static void *
application_thread( void *ptr )
{
        arm4_shm_application_instance_t app_instance;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getAppInstances (&app_instance);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processApplicationInstances (&app_instance);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "application_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "application_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
application_group_thread( void *ptr )
{
        arm4_shm_property_t app_group;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getAppGroup (&app_group);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processApplicationGroup (&app_group);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "application_group_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "application_group_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
application_instance_id_thread( void *ptr )
{
        arm4_shm_property_t property;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getAppInstanceId (&property);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processApplicationInstanceId (&property);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "application_instance_id_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "application_instance_id_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
application_context_thread( void *ptr )
{
        arm4_shm_context_property_t property;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getAppContext (&property);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processApplicationContext (&property);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "application_context_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "application_context_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
application_address_thread( void *ptr )
{
        arm4_shm_address_t address;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getAppAddress (&address);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processApplicationAddress (&address);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "application_address_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "application_address_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
transaction_thread( void *ptr )
{
        arm4_shm_transaction_instance_t tran_instance;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getTranInstances (&tran_instance);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processTransactionInstances (&tran_instance);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "transaction_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "transaction_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
transaction_context_thread( void *ptr )
{
        arm4_shm_context_property_t property;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getTranContext (&property);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processTransactionContext (&property);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "transaction_context_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "transaction_context_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
transaction_uri_thread( void *ptr )
{
        arm4_shm_uri_t uri;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getTranUri (&uri);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processTransactionUri (&uri);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "transaction_uri_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "transaction_uri_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
transaction_metric_values_thread( void *ptr )
{
        arm4_shm_metric_value_t metric;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getTranMetricValues (&metric);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processTransactionMetricValue (&metric);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "transaction_metric_values_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "transaction_metric_values_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
transaction_user_thread( void *ptr )
{
        arm4_shm_user_t user;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getTranUser (&user);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processTransactionUser (&user);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "transaction_user_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "transaction_user_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
transaction_diag_thread( void *ptr )
{
        arm4_shm_diag_t diag;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getTranDiag (&diag);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processTransactionDiag (&diag);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "transaction_diag_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "transaction_diag_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
correlator_thread( void *ptr )
{
        arm4_shm_correlator_t correlators;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getCorellators (&correlators);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processCorrelators (&correlators);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "correlator_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "correlator_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
sequence_thread( void *ptr )
{
        arm4_shm_db_sequence_t sequence;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getSequence (&sequence);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processSequence (&sequence);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "sequence_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "sequence_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
register_application_thread( void *ptr )
{
        arm4_shm_register_application_t application;
        arm_id_t appid;
        arm_boolean_t is_new;
        int status;

        /** IMPORTANT: There should only be one instance of this thread to prevent the same application being registered twice! **/

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getRegisterApplication (&application);
                        try
                        {
                                if (status > 0)
                                {
                                        status = Arm4db::getArm4db ()->processRegisterApplication (&application, &appid, &is_new);
                                        if (status == 0)
                                                Arm4dbDaemonSharedMemory::registerApplicationReturn (application.message.application_index, appid, is_new);
                                        else
                                        {
                                                /* Send bad return */
                                                memset (&appid, 0, sizeof (arm_id_t));
                                                Arm4dbDaemonSharedMemory::registerApplicationReturn (application.message.application_index, appid, ARM_FALSE);
                                        }
                                }
                                else
                                {
                                        /* Send bad return */
                                        memset (&appid, 0, sizeof (arm_id_t));
                                        Arm4dbDaemonSharedMemory::registerApplicationReturn (application.message.application_index, appid, ARM_FALSE);
                                }
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "register_application_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
        Logger::getLogger().exceptionError (e, "register_application_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
register_application_identity_thread( void *ptr )
{
        arm4_shm_register_application_identity_t identity;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getRegisterApplicationIdentity (&identity);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processRegisterApplicationIdentity (&identity);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "register_application_identity_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
        Logger::getLogger().exceptionError (e, "register_application_identity_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
register_application_context_thread( void *ptr )
{
        arm4_shm_register_application_context_t context;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getRegisterApplicationContext (&context);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processRegisterApplicationContext (&context);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "register_application_context_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
        Logger::getLogger().exceptionError (e, "register_application_context_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
register_transaction_thread( void *ptr )
{
        arm4_shm_register_transaction_t transaction;
        arm_id_t tran_id;
        arm_boolean_t is_new;
        int status;

        /** IMPORTANT: There should only be one instance of this thread to prevent the same transaction being registered twice! **/

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getRegisterTransaction (&transaction);
                        try
                        {
                                if (status > 0)
                                {
                                        status = Arm4db::getArm4db ()->processRegisterTransaction (&transaction, &tran_id, &is_new);
                                        if (status == 0)
                                                Arm4dbDaemonSharedMemory::registerTransactionReturn (transaction.message.transaction_index, tran_id, is_new);
                                        else
                                        {
                                                /* Send bad return */
                                                memset (&tran_id, 0, sizeof (arm_id_t));
                                                Arm4dbDaemonSharedMemory::registerTransactionReturn (transaction.message.transaction_index, tran_id, ARM_FALSE);
                                        }
                                }
                                else
                                {
                                        /* Send bad return */
                                        memset (&tran_id, 0, sizeof (arm_id_t));
                                        Arm4dbDaemonSharedMemory::registerTransactionReturn (transaction.message.transaction_index, tran_id, ARM_FALSE);
                                }
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "register_transaction_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "register_transaction_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
register_transaction_identity_thread( void *ptr )
{
        arm4_shm_register_application_identity_t identity;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getRegisterTransactionIdentity (&identity);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processRegisterTransactionIdentity (&identity);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "register_transaction_identity_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "register_transaction_identity_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
register_transaction_context_thread( void *ptr )
{
        arm4_shm_register_application_context_t context;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getRegisterTransactionContext (&context);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processRegisterTransactionContext (&context);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "register_transaction_context_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "register_transaction_context_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
register_transaction_uri_thread( void *ptr )
{
        arm4_shm_register_transaction_uri_t uri;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getRegisterTransactionUri (&uri);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processRegisterTransactionUri (&uri);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "register_transaction_uri_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "register_transaction_uri_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
register_transaction_metric_binding_thread( void *ptr )
{
        arm4_shm_register_transaction_metric_t metric;
        int status;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getRegisterTransactionMetricBinding (&metric);
                        try
                        {
                                if (status > 0)
                                        Arm4db::getArm4db ()->processRegisterTransactionMetricBinding (&metric);
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "register_transaction_metric_binding_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "register_transaction_metric_binding_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
register_metric_thread( void *ptr )
{
        arm4_shm_register_metric_t metric;
        arm_id_t metric_id;
        int status;

        /** IMPORTANT: There should only be one instance of this thread to prevent the same metric being registered twice! **/
        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getRegisterMetric (&metric);
                        try
                        {
                                if (status > 0)
                                {
                                        status = Arm4db::getArm4db ()->processRegisterMetric (&metric, &metric_id);
                                        if (status == 0)
                                                Arm4dbDaemonSharedMemory::registerMetricReturn (metric.message.metric_index, metric_id);
                                        else
                                        {
                                                /* Send bad return */
                                                memset (&metric_id, 0, sizeof (arm_id_t));
                                                Arm4dbDaemonSharedMemory::registerMetricReturn (metric.message.metric_index, metric_id);
                                        }
                                }
                                else
                                {
                                        /* Send bad return */
                                        memset (&metric_id, 0, sizeof (arm_id_t));
                                        Arm4dbDaemonSharedMemory::registerMetricReturn (metric.message.metric_index, metric_id);
                                }
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "register_metric_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "register_metric_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
utility_thread( void *ptr )
{
        arm4_shm_utility_t command;
        int status;

        /** IMPORTANT: There should only be one instance of this thread to prevent the same metric being registered twice! **/
        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        status = Arm4dbDaemonSharedMemory::getUtilityCommand (&command);
                        try
                        {
                                if (status > 0)
                                {
                                        switch (command.message.command)
                                        {
                                                case ARM4_SHM_COMMAND_BACKUP:
                                                        backup ();
                                                        break;
                
                                                case ARM4_SHM_COMMAND_CHECKPOINT:
                                                        checkpoint ();
                                                        break;
                
                                                case ARM4_SHM_COMMAND_STOP:
                                                        exit (0); /* Our termination handlers ensure an orderly shutdown */
                                                        break;
                
                                                default:
                                                        syslog (LOG_ERR, "arm4_daemon unknown command %ld", command.message.command);
                                        }
                                }
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "utility_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "utility_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
sample_thread( void *ptr )
{
        /* Background thread flushes the sampled traces, and ensures time sampled traces are taken */
        struct timespec requested_sleep;

        /* Request a 0.1 second sleep */
        requested_sleep.tv_sec = 0;
        requested_sleep.tv_nsec = 100000000;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        nanosleep (&requested_sleep, NULL);
                        try
                        {
                                Arm4dbDaemonSharedMemory::sampleFlush ();
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "sample_thread");
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "sample_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
stats_thread( void *ptr )
{
        arm4_shm_stats_t stats;
    DB_TXN_STAT *dbstats_ptr;

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        try
                        {
                                Arm4dbDaemonSharedMemory::getStats (&stats);
                                printf ("\nCurrent Message Queue\n");
                                printf ("Messages\tBytes\tMax Bytes\n");
                                printf ("%lu\t%lu\t%lu\n",
                                                stats.mq_n_messages,
                                                stats.mq_n_bytes,
                                                stats.mq_max_bytes);
                
                                dbstats_ptr = (DB_TXN_STAT *) Arm4db::getArm4db ()->getStats ();
                                if (dbstats_ptr)
                                {
                                        printf ("\nTransactions\n");
                                        printf ("Max %d\tActive %d\tMax Active %d\n",
                                                        dbstats_ptr->st_maxtxns,
                                                        dbstats_ptr->st_nactive,
                                                        dbstats_ptr->st_maxnactive);
                                        printf ("Begun %d\tAborted %d\tCommitted %d\tRestored %d\n",
                                                        dbstats_ptr->st_nbegins,
                                                        dbstats_ptr->st_naborts,
                                                        dbstats_ptr->st_ncommits,
                                                        dbstats_ptr->st_nrestores);
                                        free (dbstats_ptr);
                                }
                        }
                        catch (Arm4dbException e)
                        {
                                Logger::getLogger().exceptionError (e, "stats_thread");
                        }

                        arm_sleep (5);
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "stats_thread initialization");
        }

        pthread_exit (NULL);
}

static void *
db_checkpoint_thread( void *ptr )
{
        // This thread is used to maintain database consistency.
        // Transactions are checkpointed every 2 minutes, but this
        // can be changed in the config file
        int interval = Arm4dbConfig::getConfig ().getCheckpointInterval ();

        try
        {
                Arm4db::getArm4db ()->initializeThread ();
        
                for (;;)
                {
                        // A negative value indicates checkpointing is disabled
                        if (interval > 0)
                        {
                                try
                                {
                                        checkpoint ();
                                }
                                catch (Arm4dbException e)
                                {
                                        Logger::getLogger().exceptionError (e, "db_checkpoint_thread");
                                }
                                arm_sleep (interval);
                        }
                        else
                        {
                                arm_sleep (3600);
                        }
                }
        }
        catch (Arm4dbException e)
        {
                Logger::getLogger().exceptionError (e, "db_checkpoint_thread initialization");
        }

        pthread_exit (NULL);
}

static void
process_queues (void)
{
        pthread_t thread;
        int i;

        if (g_verbose)
                printf ("Creating shared memory\n");

    /* Turn collection on or off when creating the shared memory segment */
    if (g_off)
        Arm4dbDaemonSharedMemory::setNullCollector (ARM_TRUE); /* Set the library as inactive. */
    else
        Arm4dbDaemonSharedMemory::setNullCollector (ARM_FALSE); /* Set the library as active. */

        /* Create the databases if required */
        if (g_verbose)
                printf ("Opening databases\n");
        Arm4db::getArm4db ()->openDatabases ();

        atexit (on_close);

        if (g_verbose)
                printf ("Starting queue threads\n");

        /* EXPERIMENT: Launch 2 of all queues to try and improve throughput */
        for (i = 0; i < 1; i++)
                {
                pthread_create( &thread, NULL, application_thread, NULL);
                pthread_create( &thread, NULL, application_group_thread, NULL);
                pthread_create( &thread, NULL, application_instance_id_thread, NULL);
                pthread_create( &thread, NULL, application_context_thread, NULL);
                pthread_create( &thread, NULL, application_address_thread, NULL);
                pthread_create( &thread, NULL, transaction_thread, NULL);
                pthread_create( &thread, NULL, transaction_context_thread, NULL);
                pthread_create( &thread, NULL, transaction_uri_thread, NULL);
                pthread_create( &thread, NULL, transaction_metric_values_thread, NULL);
                pthread_create( &thread, NULL, transaction_user_thread, NULL);
                pthread_create( &thread, NULL, transaction_diag_thread, NULL);
                pthread_create( &thread, NULL, correlator_thread, NULL);
                pthread_create( &thread, NULL, sequence_thread, NULL);
                pthread_create( &thread, NULL, register_application_identity_thread, NULL);
                pthread_create( &thread, NULL, register_application_context_thread, NULL);
                pthread_create( &thread, NULL, register_transaction_identity_thread, NULL);
                pthread_create( &thread, NULL, register_transaction_context_thread, NULL);
                pthread_create( &thread, NULL, register_transaction_uri_thread, NULL);
                pthread_create( &thread, NULL, register_transaction_metric_binding_thread, NULL);
                }

        /* These threads should only have a single instance */
        pthread_create( &thread, NULL, register_application_thread, NULL);
        pthread_create( &thread, NULL, register_transaction_thread, NULL);
    pthread_create( &thread, NULL, register_metric_thread, NULL);
        pthread_create( &thread, NULL, utility_thread, NULL);

        /* Thread to flush sampled values, or ensure sampled values get measured */
        pthread_create( &thread, NULL, sample_thread, NULL);

        if (g_stats)
                pthread_create( &thread, NULL, stats_thread, NULL);

        if (g_verbose)
                printf ("Open for business\n");

        db_checkpoint_thread (NULL); /* Shouldn't return */
}

static void
version (void)
{
        printf ("Package %s\n", PACKAGE);
        printf ("Version %s\n", VERSION);
}

static void
usage (void)
{
        printf ("Usage:\n");
#ifdef HAVE_GETOPT_LONG
        printf ("\tarm4_daemon --help|-h\n");
        printf ("\t\tShow this help message\n");
        printf ("\tarm4_daemon --version|-V\n");
        printf ("\t\tDisplay implementation version\n");
        printf ("\tarm4_daemon [--verbose|-v] [--debug|-d] [--config|-C config_file] [--stats|-s] [--off|-o]\n");
        printf ("\t\t--verbose displays the setup progress\n");
        printf ("\t\t--debug runs the server in the foreground\n");
    printf ("\t\t--config specifies a path to the configuration file. The default is /etc/arm4.conf\n");
        printf ("\t\t--stats display the message queue stats\n");
    printf ("\t\t--off turns off all ARM 4 data collection. The daemon remains active\n");
#else
        printf ("\tarm4_daemon -h\n");
        printf ("\t\tShow this help message\n");
        printf ("\tarm4_daemon -V\n");
        printf ("\t\tDisplay implementation version\n");
        printf ("\tarm4_daemon -vdso [-C config_file]\n");
        printf ("\t\t-v displays the setup progress\n");
        printf ("\t\t-d runs the server in the foreground\n");
    printf ("\t\t-C specifies a path to the configuration file. The default is /etc/arm4.conf\n");
        printf ("\t\t-s display the message queue stats\n");
    printf ("\t\t-o turns off all ARM 4 data collection. The daemon remains active\n");
#endif
}

static void
initialize_configuration (const char *config_filename)
{
    if (g_verbose)
    {
        printf("Configuration filename '%s'\n", config_filename);
    }
        /* Set the DB parameters */
        const Arm4dbConfig &config = Arm4dbConfig::getConfig (config_filename);

        if (g_verbose)
        {
                printf ("Instance %d\n", config.getInstance ());
        }

        /* Set our user mask to allow user and group access */
    if (g_verbose)
    {
        printf("Setting umask to %04o\n", config.getUmask ());
    }
        umask (config.getUmask ());
    Arm4dbDaemonSharedMemory::setUmask (config.getUmask ());

    /* Create the database home directory */
    Arm4dbCommon::createPath (config.getDbHomeC ());
    int status = chown (config.getDbHomeC (), config.getUid (),  config.getGid ());
        if (status != 0)
        {
                perror ("chown");
                exit (1);
        }

        /* Create the IPC directory */
        if (config.getInstance () == 0)
        {
                status = Arm4dbCommon::createPath (ARM4DATA_QUEUE_DIR);
                if (status == ARM_SUCCESS)
                {
                    status = chown (ARM4DATA_QUEUE_DIR, config.getUid (),  config.getGid ());
                        if (status != 0)
                        {
                                perror ("chown");
                                exit (1);
                        }
                }
                else
                        exit (1);
        }

        /* Set our user and group - group first in case we're an unprivileged user */
        if (config.getGid () != (gid_t) -1)
        {
                if (setegid (config.getGid ()) < 0)
                {
                        perror ("setgid");
                        exit (1);
                }
        }
        if (config.getUid () != (uid_t) -1)
        {
                if (seteuid (config.getUid ()) < 0)
                {
                        perror ("setuid");
                        exit (1);
                }
        }
    // Set our umask again since our user has changed
        umask (config.getUmask ());
}

static void
daemonize (void)
{
        pid_t pid;

        if (g_debug)
        {
                if (g_verbose)
                        printf ("Running in the foreground\n");
                return;
        }

        if (g_verbose)
                printf ("Switching to background processing...\n");

        pid = fork ();

        if (pid == 0)
        {
                /* This is the child process */
                Arm4dbDaemonSharedMemory::setDaemon (ARM_TRUE); /* Replace the parent PID */
                return;
        }

        if (pid == -1)
        {
                perror ("fork");
                exit (1);
        }

        if (g_verbose)
                printf ("...pid = %d\n", pid);
        exit (0);       /* This is the parent */
}

int
main (int argc, char *argv[])
{
        int c;
        struct sigaction new_action, old_action;
    char config_filename [1024];

    // Initialize the logger
    Logger::getLogger().setName ("arm4_daemon");

    /* Set the default config file */
    strcpy (config_filename, "/etc/arm4.conf");

        while (1)
        {
#ifdef HAVE_GETOPT_LONG
                static struct option long_options[] =
                        {
                                        /* These options set a flag. */
                                {"verbose",  no_argument,       &g_verbose, 1},
                                {"brief",    no_argument,       &g_verbose, 0},
                                {"debug",    no_argument,       &g_debug, 1},
                                {"stats",    no_argument,       &g_stats, 1},
                                {"off",      no_argument,       &g_off, 1},
                                        /* These options don't set a flag. We distinguish them by their indices. */
                                {"help",     no_argument,       0, 'h'},
                                {"version",  no_argument,       0, 'V'},
                {"config",   required_argument, 0, 'C'},
                                {0, 0, 0, 0}
                        };

                /* getopt_long stores the option index here. */
                int option_index = 0;

                c = getopt_long (argc, argv, "vdhsoVC:",
                                                long_options, &option_index);
#else
                c = getopt (argc, argv, "vdhsoVC:");
#endif

                /* Detect the end of the options. */
                if (c == -1)
                        break;

                switch (c)
                {
                case 0:
                        /* If this option set a flag, do nothing else now. */
#ifdef HAVE_GETOPT_LONG
                        if (long_options[option_index].flag != 0)
                                break;

                        printf ("option %s", long_options[option_index].name);
                        if (optarg)
                                printf (" with arg %s", optarg);
                        printf ("\n");
#endif
                        break;

                case 'v':
                        g_verbose = 1;
                        break;

                case '?':
                        /* getopt_long already printed an error message. */
                        break;

                case 'd':
                        g_debug = 1;
                        break;

                case 's':
                        g_stats = 1;
                        break;

        case 'o':
            g_off = 1;
            break;

                case 'h':
                        usage ();
                        exit (0);

                case 'V':
                        version ();
                        exit (0);

        case 'C':
            strncpy (config_filename, optarg, sizeof(config_filename));
            break;

                default:
                        usage ();
                        abort ();
                }
        }

        /* Instead of reporting `--verbose'
                and `--brief' as they are encountered,
                we report the final status resulting from them. */
        if (g_verbose)
    {
                puts ("verbose flag is set");
        Logger::getLogger().setFile (stderr);
    }

    initialize_configuration (config_filename);

        if (!g_verbose)
        {
                Logger::getLogger().setFile (Arm4dbConfig::getConfig ().getLogFilename ());
        }

        // Make sure the IPC structures are created if they don't already exist
        //  and start as a null collector
        Arm4dbDaemonSharedMemory::shmCreate (ARM_TRUE);

        pid_t daemon_pid = Arm4dbDaemonSharedMemory::getActiveDaemon ();
        if ((daemon_pid != 0) && (daemon_pid != getpid()))
        {
                fprintf (stderr, "arm4_daemon is already running\n");
                exit (1);
        }
        Arm4dbDaemonSharedMemory::setDaemon (ARM_FALSE); /* Claim the daemon for now */

        if (g_verbose)
                puts ("Setting signal handlers");

    /* Set up the structure to specify the new action. */
    new_action.sa_handler = termination_handler;
    sigemptyset (&new_action.sa_mask);
    new_action.sa_flags = 0;

    sigaction (SIGINT, NULL, &old_action);
    if (old_action.sa_handler != SIG_IGN)
                sigaction (SIGINT, &new_action, NULL);
    sigaction (SIGTERM, NULL, &old_action);
    if (old_action.sa_handler != SIG_IGN)
                sigaction (SIGTERM, &new_action, NULL);
        sigaction (SIGHUP, NULL, &old_action);
        if (old_action.sa_handler != SIG_IGN)
                sigaction (SIGHUP, &new_action, NULL);

        /* Print any remaining command line arguments (not options). */
        if (optind < argc)
        {
                printf ("non-option ARGV-elements: ");
                while (optind < argc)
                        printf ("%s ", argv[optind++]);
                putchar ('\n');
        }

        daemonize ();
        process_queues (); /* Should not return */

        /* Stop the main thread allowing the children to continue */
        pthread_exit (NULL);
}