///////////////////////////////////////////////////////////////////////////////
//
// @@@ START COPYRIGHT @@@
//
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.
//
// @@@ END COPYRIGHT @@@
//
///////////////////////////////////////////////////////////////////////////////

#include <iostream>
#include <map>

using namespace std;

#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <fcntl.h>
#include <unistd.h>
#include <malloc.h>
#include <sys/ipc.h>
#include <sys/time.h>
#include <sys/resource.h> //add for getrlimit, strange centos6/gcc4.4 don't need this

//#include <sys/stat.h>
#include "monlogging.h"
#include "monprof.h"
#include "monsonar.h"
#include "montrace.h"
#include "redirector.h"
#include "healthcheck.h"
#include "config.h"
#include "device.h"
#include "clusterconf.h"
#include "nameserverconfig.h"
#include "lnode.h"
#include "pnode.h"
#include "tmsync.h"
#include "cluster.h"
#include "monitor.h"
#include "props.h"

#ifdef DMALLOC
#include "dm.h"
#endif
#include "replicate.h"
#include "robsem.h"
#include "commaccept.h"
#ifdef NAMESERVER_PROCESS
#include "nscommacceptmon.h"
#endif

#include <assert.h>
#include <signal.h>
#include <sys/ptrace.h>
#include <sys/time.h>
#include <sys/wait.h>
#include <sched.h>
#include "localio.h"
#include "lock.h"
#include "mlio.h"
#include "redirector.h"
#include "intprocess.h"
#include "nameserver.h"
#include "meas.h"

#include "reqqueue.h"
#include "reqworker.h"
#include "zclient.h"

#include "SCMVersHelp.h"

#ifndef NAMESERVER_PROCESS
#include "ptpcommaccept.h"
#include "ptpclient.h"
#endif

#define LOG_ERROR
#define RidRecvPrioritySlot 4096
#define MinimumSlotToPrioritize 4096


// Global Variables
struct rlimit Rl;
bool PidMap = false;
bool usingCpuAffinity = false;
bool usingTseCpuAffinity = false;
bool genSnmpTrapEnabled = false;
int Measure = 0;
long trace_level = 0;
char MyPath[MAX_PROCESS_PATH];
char MyCommPort[MPI_MAX_PORT_NAME] = {'\0'};
char MyMPICommPort[MPI_MAX_PORT_NAME] = {'\0'};
char MySyncPort[MPI_MAX_PORT_NAME] = {'\0'};
#ifdef NAMESERVER_PROCESS
char MyMon2NsPort[MPI_MAX_PORT_NAME] = {'\0'};
#else
char MyPtPPort[MPI_MAX_PORT_NAME] = {'\0'};
#endif
char Node_name[MPI_MAX_PROCESSOR_NAME] = {'\0'};
sigset_t SigSet;
bool Emulate_Down = false;
#ifdef NAMESERVER_PROCESS
long next_test_delay = 10000; // in usec. (default 10 msec)
#else
long next_test_delay = 100000; // in usec. (default 100 msec)
#endif
CClusterConfig *ClusterConfig = NULL;
bool IAmIntegrating = false;
bool IAmIntegrated = false;
char IntegratingMonitorPort[MPI_MAX_PORT_NAME] = {'\0'};
bool IsRealCluster = true;
bool IsAgentMode = false;
bool IsNameServer = false;
bool IsMaster = false;
bool IsMPIChild = false;
char MasterMonitorName[MAX_PROCESS_PATH]= {'\0'};
CommType_t CommType = CommType_Undefined;
bool SMSIntegrating = false;
int  CreatorShellPid = -1;
Verifier_t CreatorShellVerifier = -1;
bool SpareNodeColdStandby = true;
bool ZClientEnabled = true;
#ifndef NAMESERVER_PROCESS
bool NameServerEnabled = false;
#endif

// Lock to manage memory modifications during fork/exec
CLock MemModLock;
CMonitor *Monitor = NULL;
#ifndef NAMESERVER_PROCESS
CNameServer *NameServer = NULL;
CProcess *NameServerProcess = NULL;
CPtpClient *PtpClient = NULL;
#endif
CNodeContainer *Nodes = NULL;
CConfigContainer *Config = NULL;
CNameServerConfigContainer *NameServerConfig = NULL;
#ifndef NAMESERVER_PROCESS
CDeviceContainer *Devices = NULL;
#endif
int MyPNID = -1;
CNode *MyNode;
CMonLog *MonLog =  NULL;
CMonStats * MonStats = NULL;
extern CMonTrace *MonTrace;
#ifndef NAMESERVER_PROCESS
CRedirector Redirector;
#endif
CIntProcess IntProcess;
CReqQueue ReqQueue;
CHealthCheck HealthCheck;
CCommAccept CommAccept;
#ifdef NAMESERVER_PROCESS
CCommAcceptMon CommAcceptMon;
#else
CPtpCommAccept PtpCommAccept;
#endif
extern CReplicate Replicator;
CZClient  *ZClient = NULL;
#ifndef NAMESERVER_PROCESS
// Seabed disconnect semaphore
RobSem * sbDiscSem = NULL;
#endif
int monitorArgc = 0;
char monitorArgv[MAX_ARGS][MAX_ARG_SIZE];
CMeas Meas;


#ifdef NAMESERVER_PROCESS
DEFINE_EXTERN_COMP_DOVERS(trafns)
DEFINE_EXTERN_COMP_GETVERS2(trafns)
#else
DEFINE_EXTERN_COMP_DOVERS(monitor)
DEFINE_EXTERN_COMP_GETVERS2(monitor)
#endif


_TM_Txid_External invalid_trans( void )
{
    _TM_Txid_External trans1;

    trans1.txid[0] = -1LL;
    trans1.txid[1] = -1LL;
    trans1.txid[2] = -1LL;
    trans1.txid[3] = -1LL;

    return trans1;
}

_TM_Txid_External null_trans( void )
{
    _TM_Txid_External trans1;

    trans1.txid[0] = 0LL;
    trans1.txid[1] = 0LL;
    trans1.txid[2] = 0LL;
    trans1.txid[3] = 0LL;

    return trans1;
}

bool isEqual( _TM_Txid_External trans1, _TM_Txid_External trans2 )
{
    return (memcmp(&trans1,&trans2,sizeof(_TM_Txid_External)) == 0);
}

bool isNull( _TM_Txid_External transid )
{
    _TM_Txid_External trans_null = null_trans();

    return isEqual(transid,trans_null);
}

bool isInvalid( _TM_Txid_External transid )
{
    _TM_Txid_External trans_invalid = invalid_trans();

    return isEqual(transid,trans_invalid);
}

char *ErrorMsg (int error_code)
{
    int rc;
    int length;
    static char buffer[MPI_MAX_ERROR_STRING];

    rc = MPI_Error_string (error_code, buffer, &length);
    if (rc != MPI_SUCCESS)
    {
        snprintf(buffer, sizeof(buffer),
                 "MPI_Error_string: Invalid error code (%d)\n", error_code);
        length = strlen(buffer);
    }
    buffer[length] = '\0';

    return buffer;
}

#ifndef NAMESERVER_PROCESS
void child_death_signal_handler2 (int signal, siginfo_t *info, void *)
{
    pid_t pid;
    int saveerrno;
    int status;
    pid_t whichPid;

    const char method_name[] = "child_death_signal_handler2";
    if (trace_settings & TRACE_ENTRY_EXIT)
       trace_nolock_printf("%s@%d\n", method_name, __LINE__);

    saveerrno = errno; // waitpid/etc sets errno

    if (trace_settings & TRACE_SIG_HANDLER)
        trace_nolock_printf("%s@%d - signal=%d, code=%d, status=%d, pid=%d\n",
                            method_name, __LINE__, signal, info->si_code,
                            info->si_status, info->si_pid);

    // Handle the process that triggered the signal handler as well
    // as any other exited children.
    whichPid = info->si_pid;
    do
    {
        pid = waitpid (whichPid, &status, WNOHANG);
        if (trace_settings & TRACE_SIG_HANDLER)
        {
            if ( pid != -1 )
            {
                trace_nolock_printf("%s@%d - waitpid(%d) returned %d\n",
                                    method_name, __LINE__, whichPid, pid);
            }
            else
            {
                trace_nolock_printf("%s@%d - waitpid(%d) error, %s (%d)\n",
                                    method_name, __LINE__, whichPid,
                                    strerror(errno), errno);
            }
        }
        if (pid > 0)
        {
            IntProcess.handle_signal(pid, status);

            // Add this pid to a list to be examined outside the
            // signal-handler (to avoid spending excessive time in
            // signal handler).  When that list is examined the work
            // required to handle the terminated process will be done.
            SQ_theLocalIOToClient->handleDeadPid(pid);

            if (trace_settings & TRACE_SIG_HANDLER)
            {
                if ( WIFEXITED(status) )
                {   // Process exited normally
                    trace_nolock_printf("%s@%d - process %d exited, exit"
                                        " status=%d\n", method_name,__LINE__,
                                        pid, WEXITSTATUS(status));
                }
                if ( WIFSIGNALED(status) )
                {   // Process was terminated by a signal
                   trace_nolock_printf("%s@%d - process %d terminated by "
                                       "signal #%d\n", method_name, __LINE__,
                                       pid, WTERMSIG(status));
                }
            }
        }
        else if ( whichPid != -1 && pid == 0 )
        {
            // Process that triggered the signal handler has not yet changed
            // state.  Remember the "pid" so we can handle it later.
            if (trace_settings & TRACE_SIG_HANDLER)
                trace_nolock_printf("%s@%d - adding pid=%d to unreaped list\n",
                                    method_name, __LINE__, pid);
            SQ_theLocalIOToClient->handleAlmostDeadPid(whichPid);
            pid = 1; // force at least one more waitpid to be done
        }
        whichPid = -1; // now do waitpid on any child process
    }
    while (pid > 0);

    errno = saveerrno;

    if (trace_settings & TRACE_ENTRY_EXIT)
        trace_nolock_printf("%s@%d - Exit\n", method_name, __LINE__);
}
#endif

void monMallocStats()
{
    // Log current malloc statistics in stderr
    time_t mytime = time(NULL);
    char *timestamp = ctime(&mytime);
    timestamp[strlen(timestamp)-1] = '\0';

    printf("monitor malloc statistics at %s:\n", timestamp);
    malloc_stats();
}

const char *CommTypeString( CommType_t commType)
{
    const char *str;

    switch( commType )
    {
        case CommType_InfiniBand:
            str = "InfiniBand";
            break;
        case CommType_Sockets:
            str = "Sockets";
            break;
        default:
            str = "Undefined";
            break;
    }

    return( str );
}


#ifdef NAMESERVER_PROCESS
CMonitor::CMonitor ()
    : CCluster (),
#else
CMonitor::CMonitor (int procTermSig)
    : CTmSync_Container (),
#endif
      OpenCount (0)
    , NoticeCount (0)
    , ProcessCount (0)
    , NumOutstandingIO (0)
    , NumOutstandingSends (0)
    , Last_error (MPI_SUCCESS)
#ifndef NAMESERVER_PROCESS
    , processMapFd ( -1 )
    , procTermSig_ ( procTermSig )
#endif
{
    const char method_name[] = "CMonitor::CMonitor";
    TRACE_ENTRY;

    // Add eyecatcher sequence as a debugging aid
    memcpy(&eyecatcher_, "MNTR", 4);

#ifdef USE_SEQUENCE_NUM
    clock_gettime(CLOCK_REALTIME, &savedTime_);
#endif

    TRACE_EXIT;
}

CMonitor::~CMonitor (void)
{
    const char method_name[] = "CMonitor::~CMonitor";
    TRACE_ENTRY;

    // Alter eyecatcher sequence as a debugging aid to identify deleted object
    memcpy(&eyecatcher_, "mntr", 4);

#ifndef NAMESERVER_PROCESS
    if ( processMapFd != -1)
    {
        close ( processMapFd );
    }
#endif

    TRACE_EXIT;
}

void CMonitor::IncOpenCount (void)
{
    OpenCount++;
}
void CMonitor::IncNoticeCount (void)
{
    NoticeCount++;
}
void CMonitor::IncProcessCount (void)
{
    ProcessCount++;
}
void CMonitor::DecrOpenCount (void)
{
    OpenCount--;
}
void CMonitor::DecrNoticeCount (void)
{
    NoticeCount--;
}
void CMonitor::DecrProcessCount (void)
{
    ProcessCount--;
}

#ifndef NAMESERVER_PROCESS
void CMonitor::openProcessMap ( void )
{
    char fname[MAX_PROCESS_PATH];

    char *env;
    env = getenv("SQ_PIDMAP");
    if ( env && *env == '1' )
    {
        PidMap = true;
    }

    snprintf( fname, sizeof(fname), "%s/monitor.map.%d.%s",
             getenv("MPI_TMPDIR"), MyPNID, Node_name );
    remove(fname);
    processMapFd = open(fname, O_WRONLY | O_APPEND | O_CREAT,
                        S_IRUSR | S_IWUSR );
    if ( processMapFd == -1 )
    {  // File open error
        char buf[MON_STRING_BUF_SIZE];
        snprintf(buf, sizeof(buf),
                 "[CMonitor::openProcessMap], Error opening %s, %s (%d).\n",
                fname, strerror(errno), errno);
        mon_log_write(MON_PROCESS_COMPLETEPSTARTUP_2, SQ_LOG_ERR, buf);
    }
}
#endif

#ifndef NAMESERVER_PROCESS
void CMonitor::writeProcessMapEntry ( const char * buf )
{
    if ( processMapFd != -1 )
        write( processMapFd, buf, strlen(buf));
}
#endif

#ifndef NAMESERVER_PROCESS
void CMonitor::writeProcessMapBegin( const char *name
                                   , int nid
                                   , int pid
                                   , int verifier
                                   , int parentNid
                                   , int parentPid
                                   , int parentVerifier
                                   , const char *program )
{
    char buf[55+MAX_PROCESS_NAME+MAX_PROCESS_PATH];

    time_t mytime = time(NULL);
    char *timestamp = ctime(&mytime);
    timestamp[strlen(timestamp)-1] = '\0';

    snprintf( buf, sizeof(buf)
            , "BEGIN %s %-8s (%d, %d:%d) P(%d, %d:%d) %s\n"
            , timestamp, name, nid, pid, verifier
            , parentNid, parentPid, parentVerifier, program);
    writeProcessMapEntry ( buf );
}
#endif

#ifndef NAMESERVER_PROCESS
void CMonitor::writeProcessMapEnd( const char *name
                                 , int nid
                                 , int pid
                                 , int verifier
                                 , int parentNid
                                 , int parentPid
                                 , int parentVerifier
                                 , const char *program )
{
    char buf[55+MAX_PROCESS_NAME+MAX_PROCESS_PATH];

    time_t mytime = time(NULL);
    char *timestamp = ctime(&mytime);
    timestamp[strlen(timestamp)-1] = '\0';

    snprintf( buf, sizeof(buf)
            , "END   %s %-8s (%d, %d:%d) P(%d, %d:%d) %s\n"
            , timestamp, name, nid, pid, verifier
            , parentNid, parentPid, parentVerifier, program);
    writeProcessMapEntry ( buf );
}
#endif

bool CMonitor::CompleteProcessStartup (struct message_def * msg)
{
    bool status = FAILURE;
    CProcess *process;
    CLNode  *lnode;

    const char method_name[] = "CMonitor::CompleteProcessStartup";
    TRACE_ENTRY;

    lnode = Nodes->GetLNode( msg->u.request.u.startup.nid );
    if ( lnode )
    {
        process = lnode->
            CompleteProcessStartup ( msg->u.request.u.startup.process_name,
                                     msg->u.request.u.startup.port_name,
                                     msg->u.request.u.startup.os_pid,
                                     msg->u.request.u.startup.event_messages,
                                     msg->u.request.u.startup.system_messages,
                                     NULL,
                                     -1 );
    }
    else
    {
        char buf[MON_STRING_BUF_SIZE];
        snprintf(buf, sizeof(buf), "[%s], Invalid Node ID=%d\n",
                 method_name, msg->u.request.u.startup.nid);
        mon_log_write(MON_MONITOR_COMPLETEPSTARTUP_1, SQ_LOG_ERR, buf);

        process = NULL;
    }

    if (process)
    {
        if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS))
        {
            trace_printf("%s@%d - Process %s started on port %s\n",
                         method_name, __LINE__,
                         msg->u.request.u.startup.process_name,
                         msg->u.request.u.startup.port_name);
        }
#ifndef NAMESERVER_PROCESS
        CProcessContainer::ParentNewProcReply( process, MPI_SUCCESS);
        status = SUCCESS;
#endif
    }
    else
    {
        char buf[MON_STRING_BUF_SIZE];
        snprintf(buf, sizeof(buf), "[CMonitor::CompleteProcessStartup], Error= Can't find process: %s!\n", msg->u.request.u.startup.process_name);
        mon_log_write(MON_MONITOR_COMPLETEPSTARTUP_2, SQ_LOG_ERR, buf);

        msg->u.reply.type = ReplyType_Generic;
        msg->u.reply.u.generic.nid = -1;
        msg->u.reply.u.generic.pid = -1;
        msg->u.reply.u.generic.verifier = -1;
        msg->u.reply.u.generic.process_name[0] = '\0';
        msg->u.reply.u.generic.return_code = MPI_ERR_NAME;
        status = FAILURE;
    }
    TRACE_EXIT;

    return status;
}

char * CMonitor::ProcCopy(char *bufPtr, CProcess *process)
{
    const char method_name[] = "CMonitor::ProcCopy";
    TRACE_ENTRY;

    int  stringDataLen = 0;
    struct clone_def *procObj = (struct clone_def *)bufPtr;

    procObj->nid = process->GetNid();
    procObj->type = process->GetType();
    procObj->priority = process->GetPriority();
    procObj->backup = process->IsBackup();
    procObj->unhooked = process->IsUnhooked();
#ifndef NAMESERVER_PROCESS
    procObj->pathStrId = process->pathStrId();
    procObj->ldpathStrId = process->ldPathStrId();
    procObj->programStrId = process->programStrId();
#endif
    procObj->os_pid = process->GetPid();
    procObj->verifier = process->GetVerifier();
    procObj->prior_pid = process->GetPriorPid ();
    procObj->parent_nid = process->GetParentNid();
    procObj->parent_pid = process->GetParentPid();
    procObj->parent_verifier = process->GetParentVerifier();
    procObj->persistent_retries = process->GetPersistentRetries();
    procObj->event_messages = process->IsEventMessages();
    procObj->system_messages = process->IsSystemMessages();
    procObj->argc = process->argc();
    procObj->creation_time = process->GetCreationTime();


    if (trace_settings & (TRACE_REQUEST | TRACE_INIT | TRACE_RECOVERY))
            trace_printf( "%s@%d - Packing process %s (%d,%d:%d)\n"
                        , method_name, __LINE__
                        , process->GetName()
                        , process->GetNid()
                        , process->GetPid()
                        , process->GetVerifier() );

    char *stringData = &procObj->stringData;

    if (strlen(process->GetName()))
    {
        // Copy the program name
        procObj->nameLen = strlen(process->GetName()) + 1;
        memcpy(stringData, process->GetName(),  procObj->nameLen );
        stringData += procObj->nameLen;
        stringDataLen = procObj->nameLen;
    }
    else
    {
        procObj->nameLen = 0;
    }

    if (strlen(process->GetPort()))
    {
        // Copy the port
        procObj->portLen = strlen(process->GetPort()) + 1;
        memcpy(stringData, process->GetPort(),  procObj->portLen );
        stringData += procObj->portLen;
        stringDataLen += procObj->portLen;
    }
    else
    {
        procObj->portLen = 0;
    }

#ifdef NAMESERVER_PROCESS
    if (strlen(process->path()))
    {
        // Copy the path
        procObj->pathLen = strlen(process->path()) + 1;
        memcpy(stringData, process->path(),  procObj->pathLen );
        stringData += procObj->pathLen;
        stringDataLen = procObj->pathLen;
    }
    else
    {
        procObj->pathLen = 0;
    }

    if (strlen(process->ldpath()))
    {
        // Copy the ldpath
        procObj->ldpathLen = strlen(process->ldpath()) + 1;
        memcpy(stringData, process->ldpath(),  procObj->ldpathLen );
        stringData += procObj->ldpathLen;
        stringDataLen = procObj->ldpathLen;
    }
    else
    {
        procObj->ldpathLen = 0;
    }

    if (strlen(process->program()))
    {
        // Copy the program
        procObj->programLen = strlen(process->program()) + 1;
        memcpy(stringData, process->program(),  procObj->programLen );
        stringData += procObj->programLen;
        stringDataLen = procObj->programLen;
    }
    else
    {
        procObj->programLen = 0;
    }
#endif
    
    if (process->IsPersistent())
    {
        if (strlen(process->infile()))
        {
            // Copy the standard in file name
            procObj->infileLen = strlen(process->infile()) + 1;
            memcpy(stringData, process->infile(), procObj->infileLen);
            stringData += procObj->infileLen;
            stringDataLen += procObj->infileLen;
        }
        else
        {
            procObj->infileLen = 0;
        }

        if (strlen(process->outfile()))
        {
            // Copy the standard out file name
            procObj->outfileLen = strlen(process->outfile()) + 1;
            memcpy(stringData, process->outfile(),  procObj->outfileLen );
            stringData += procObj->outfileLen;
            stringDataLen += procObj->outfileLen;
        }
        else
        {
            procObj->outfileLen = 0;
        }

        procObj->argvLen =  process->userArgvLen();
        if (procObj->argvLen)
        {
            // Copy the program argument strings
            memcpy(stringData, process->userArgv(), procObj->argvLen);
            stringData += procObj->argvLen;
            stringDataLen += procObj->argvLen;
        }

        procObj->persistent = true;
    }
    else
    {
        procObj->infileLen = 0;
        procObj->outfileLen = 0;
        procObj->argvLen = 0;
        procObj->persistent = false;
    }

#ifdef NAMESERVER_PROCESS
    if (trace_settings & (TRACE_REQUEST | TRACE_INIT | TRACE_RECOVERY))
            trace_printf( "%s@%d - Packing process string data:\n"
                          "        name(%d)       =%s\n"
                          "        port(%d)       =%s\n"
                          "        path(%d)       =%s\n"
                          "        ldpath(%d)     =%s\n"
                          "        program(%d)    =%s\n"
                          "        infile(%d)     =%s\n"
                          "        outfile(%d)    =%s\n"
                          "        userArgv(%d)   =%s\n"
                          "        stringData(%d) =%s\n"
                        , method_name, __LINE__
                        , procObj->nameLen
                        , process->GetName()
                        , procObj->portLen
                        , process->GetPort()
                        , procObj->pathLen
                        , process->path()
                        , procObj->ldpathLen
                        , process->ldpath()
                        , procObj->programLen
                        , process->program()
                        , procObj->infileLen
                        , process->infile()
                        , procObj->outfileLen
                        , process->outfile()
                        , procObj->argvLen
                        , procObj->argvLen?process->userArgv():""
                        , stringDataLen
                        , stringDataLen?&procObj->stringData:"" );
#else
    if (trace_settings & (TRACE_REQUEST | TRACE_INIT | TRACE_RECOVERY))
            trace_printf( "%s@%d - Packing process string data:\n"
                          "        name(%d)       =%s\n"
                          "        port(%d)       =%s\n"
                          "        infile(%d)     =%s\n"
                          "        outfile(%d)    =%s\n"
                          "        userArgv(%d)   =%s\n"
                          "        stringData(%d) =%s\n"
                        , method_name, __LINE__
                        , procObj->nameLen
                        , process->GetName()
                        , procObj->portLen
                        , process->GetPort()
                        , procObj->infileLen
                        , process->infile()
                        , procObj->outfileLen
                        , process->outfile()
                        , procObj->argvLen
                        , procObj->argvLen?process->userArgv():""
                        , stringDataLen
                        , stringDataLen?&procObj->stringData:"" );
#endif

    TRACE_EXIT;
    return stringData;
}

int CMonitor::PackProcObjs( char *&buffer )
{
    const char method_name[] = "CMonitor::PackProcObjs";
    TRACE_ENTRY;

    CLNode *lnode = NULL;
    CProcess *process = NULL;
    int procCount = 0;

    char *bufPtr = buffer;

#ifndef NAMESERVER_PROCESS
    if (!NameServerEnabled)
#endif
    {
        // first copy all primary and generic processes
        lnode = Nodes->GetFirstLNode();
        for ( ; lnode ; lnode = lnode->GetNext() )
        {
            process = lnode->GetFirstProcess();
            while (process)
            {
                if (!process->IsBackup())
                {
                    buffer = ProcCopy(buffer, process);
                    ++procCount;
                }
    
                process = process->GetNext();
            }
        }
    
        // copy all the backup processes
        lnode = Nodes->GetFirstLNode();
        for ( ; lnode ; lnode = lnode->GetNext() )
        {
            process = lnode->GetFirstProcess();
            while (process)
            {
                if (process->IsBackup())
                {
                    buffer = ProcCopy(buffer, process);
                    ++procCount;
                }
    
                process = process->GetNext();
            }
        }
    }

    if (trace_settings & (TRACE_REQUEST | TRACE_INIT | TRACE_RECOVERY))
    {
        long avg = 0;
        if ( procCount > 0 )
            avg = (buffer - bufPtr) / procCount;
        trace_printf("%s@%d - Total Procs = %d, Total Size = %ld, Avg = %ld bytes\n",
                     method_name, __LINE__, procCount, buffer - bufPtr, avg);
    }

    TRACE_EXIT;
    return procCount;
}

void CMonitor::UnpackProcObjs( char *&buffer, int procCount )
{
    const char method_name[] = "CMonitor::UnpackProcObjs";
    TRACE_ENTRY;

    CNode * node = NULL;
    CProcess * process = NULL;
    int  stringDataLen;
#ifdef NAMESERVER_PROCESS
    char *path = NULL;
    char *ldpath = NULL;
    char *program = NULL;
#endif
    char *name = NULL;
    char *port = NULL;
    char *infile = NULL;
    char *outfile = NULL;
    char *userargv = NULL;
    char *stringData = NULL;

    struct clone_def *procObj;

    int i;

    for (i = 0; i < procCount; i++)
    {
        procObj = (struct clone_def *)buffer;

        stringDataLen = 0;
        stringData = &procObj->stringData;

        node = Nodes->GetLNode (procObj->nid)->GetNode();

        if (procObj->nameLen)
        {
            name = &stringData[stringDataLen];
            stringDataLen += procObj->nameLen;
        }

        if (procObj->portLen)
        {
            port = &stringData[stringDataLen];
            stringDataLen += procObj->portLen;
        }

#ifdef NAMESERVER_PROCESS
        if (procObj->pathLen)
        {
            path = &stringData[stringDataLen];
            stringDataLen += procObj->pathLen;
        }

        if (procObj->ldpathLen)
        {
            ldpath = &stringData[stringDataLen];
            stringDataLen += procObj->ldpathLen;
        }

        if (procObj->programLen)
        {
            program = &stringData[stringDataLen];
            stringDataLen += procObj->programLen;
        }
#endif

        if (procObj->infileLen)
        {
            infile = &stringData[stringDataLen];
            stringDataLen += procObj->infileLen;
        }

        if (procObj->outfileLen)
        {
            outfile = &stringData[stringDataLen];
            stringDataLen += procObj->outfileLen;
        }

        if (procObj->argvLen)
        {
            userargv = &stringData[stringDataLen];
            stringDataLen += procObj->argvLen;
        }

#ifdef NAMESERVER_PROCESS
        if (trace_settings & (TRACE_REQUEST | TRACE_INIT | TRACE_RECOVERY))
                trace_printf( "%s@%d - Unpacking process string data:\n"
                              "        stringData(%d) =%s\n"
                              "        name(%d)       =%s\n"
                              "        port(%d)       =%s\n"
                              "        path(%d)       =%s\n"
                              "        ldpath(%d)     =%s\n"
                              "        program(%d)    =%s\n"
                              "        infile(%d)     =%s\n"
                              "        outfile(%d)    =%s\n"
                              "        userArgc       =%d\n"
                              "        userArgv(%d)   =%s\n"
                            , method_name, __LINE__
                            , stringDataLen
                            , stringDataLen?&procObj->stringData:""
                            , procObj->nameLen
                            , procObj->nameLen?name:""
                            , procObj->portLen
                            , procObj->portLen?port:""
                            , procObj->pathLen
                            , procObj->pathLen?path:""
                            , procObj->ldpathLen
                            , procObj->ldpathLen?ldpath:""
                            , procObj->programLen
                            , procObj->programLen?program:""
                            , procObj->infileLen
                            , procObj->infileLen?infile:""
                            , procObj->outfileLen
                            , procObj->outfileLen?outfile:""
                            , procObj->argc
                            , procObj->argvLen
                            , procObj->argvLen?userargv:"" );
#else
        if (trace_settings & (TRACE_REQUEST | TRACE_INIT | TRACE_RECOVERY))
                trace_printf( "%s@%d - Unpacking process string data:\n"
                              "        stringData(%d) =%s\n"
                              "        name(%d)       =%s\n"
                              "        port(%d)       =%s\n"
                              "        infile(%d)     =%s\n"
                              "        outfile(%d)    =%s\n"
                              "        userArgc       =%d\n"
                              "        userArgv(%d)   =%s\n"
                            , method_name, __LINE__
                            , stringDataLen
                            , stringDataLen?&procObj->stringData:""
                            , procObj->nameLen
                            , procObj->nameLen?name:""
                            , procObj->portLen
                            , procObj->portLen?port:""
                            , procObj->infileLen
                            , procObj->infileLen?infile:""
                            , procObj->outfileLen
                            , procObj->outfileLen?outfile:""
                            , procObj->argc
                            , procObj->argvLen
                            , procObj->argvLen?userargv:"" );
#endif

        process = node->CloneProcess (procObj->nid,
                                      procObj->type,
                                      procObj->priority,
                                      procObj->backup,
                                      procObj->unhooked,
                                      procObj->nameLen?name:(char *)"",
                                      procObj->portLen?port:(char *)"",
                                      procObj->os_pid,
                                      procObj->verifier,
                                      procObj->parent_nid,
                                      procObj->parent_pid,
                                      procObj->parent_verifier,
                                      procObj->event_messages,
                                      procObj->system_messages,
#ifdef NAMESERVER_PROCESS
                                      path,
                                      ldpath,
                                      program,
#else
                                      procObj->pathStrId,
                                      procObj->ldpathStrId,
                                      procObj->programStrId,
#endif
                                      procObj->infileLen?infile:(char *)"",
                                      procObj->outfileLen?outfile:(char *)"",
                                      &procObj->creation_time,
                                      procObj->origPNidNs);

        if ( process && procObj->argvLen )
        {
            process->userArgs ( procObj->argc, procObj->argvLen, userargv );
        }

        if ( process && procObj->persistent )
        {
            process->SetPersistent(true);
        }

        buffer = &stringData[stringDataLen];
    }

    TRACE_EXIT;
    return;
}

#ifndef NAMESERVER_PROCESS
void CMonitor::StartPrimitiveProcesses( void )
{
    const char method_name[] = "CMonitor::StartPrimitiveProcesses";
    TRACE_ENTRY;

    if ( !MyNode->IsSpareNode() )
    {
        // Queue the Create primitive processes request for
        // processing by a worker thread.
        ReqQueue.enqueueCreatePrimitiveReq( MyPNID );
    }

    TRACE_EXIT;
}
#endif

void HandleAssignMonitorLeader ( const char *failedMaster )
{
    const char method_name[] = "HandleAssignMonitorLeader";
    TRACE_ENTRY;
    if (trace_settings & (TRACE_INIT | TRACE_RECOVERY))
    {
        trace_printf("%s@%d HandleAssignMonitorLeader called for %s\n"
                            , method_name, __LINE__, failedMaster);
    }
    // only relevant in AgentMode
     if (IsAgentMode)
     {
         Monitor->AssignMonitorLeader(failedMaster);
     }
    
    TRACE_EXIT;
}

void HandleMyNodeExpiration( void )
{
    const char method_name[] = "HandleMyNodeExpiration";
    TRACE_ENTRY;
    ReqQueue.enqueueDownReq(MyPNID);
    TRACE_EXIT;
}

void HandleNodeExpiration( const char *nodeName )
{
    const char method_name[] = "HandleNodeExpiration";
    TRACE_ENTRY;
    CNode *node = Nodes->GetNode((char *)nodeName);
    if (node)
    {
        ReqQueue.enqueueDownReq(node->GetPNid());
    }
    TRACE_EXIT;
}

void CreateZookeeperClient( void )
{
    const char method_name[] = "CreateZookeeperClient";
    TRACE_ENTRY;

    if ( ZClientEnabled )
    {
        string       hostName;
        string       zkQuorumHosts;
        stringstream zkQuorumPort;
        char *env;
        char  hostsStr[MAX_PROCESSOR_NAME * 3] = { 0 };
        char *tkn = NULL;

        int zport;
        env = getenv("ZOOKEEPER_PORT");
        if ( env && isdigit(*env) )
        {
            zport = atoi(env);
        }
        else
        {
            char buf[MON_STRING_BUF_SIZE];
            snprintf(buf, sizeof(buf),
                     "[%s], Zookeeper quorum port is not defined!\n"
                    , method_name);
            mon_log_write(MON_MONITOR_CREATEZCLIENT_1, SQ_LOG_CRIT, buf);

            ZClientEnabled = false;
            TRACE_EXIT;
            return;
        }

        env = getenv("ZOOKEEPER_NODES");
        if ( env )
        {
            zkQuorumHosts = env;
            if ( zkQuorumHosts.length() == 0 )
            {
                char buf[MON_STRING_BUF_SIZE];
                snprintf(buf, sizeof(buf),
                         "[%s], Zookeeper quorum hosts are not defined!\n"
                        , method_name);
                mon_log_write(MON_MONITOR_CREATEZCLIENT_2, SQ_LOG_CRIT, buf);

                ZClientEnabled = false;
                TRACE_EXIT;
                return;
            }

            strcpy( hostsStr, zkQuorumHosts.c_str() );
            zkQuorumPort.str( "" );

            tkn = strtok( hostsStr, "," );
            do
            {
                if ( tkn != NULL )
                {
                    hostName = tkn;
                    zkQuorumPort << hostName.c_str()
                                 << ":"
                                 << zport;
                }
                tkn = strtok( NULL, "," );
                if ( tkn != NULL )
                {
                    zkQuorumPort << ",";
                }

            }
            while( tkn != NULL );
            if (trace_settings & (TRACE_INIT | TRACE_RECOVERY))
            {
                trace_printf( "%s@%d zkQuorumPort=%s\n"
                            , method_name, __LINE__
                            , zkQuorumPort.str().c_str() );
            }
        }

        ZClient = new CZClient( zkQuorumPort.str().c_str()
                              , ZCLIENT_TRAFODION_ZNODE
                              , ZCLIENT_INSTANCE_ZNODE );
        if ( ZClient == NULL )
        {
            char buf[MON_STRING_BUF_SIZE];
            snprintf(buf, sizeof(buf),
                     "[%s], Failed to allocate ZClient object!\n"
                    , method_name);
            mon_log_write(MON_MONITOR_CREATEZCLIENT_3, SQ_LOG_CRIT, buf);
            abort();
        }
    }

    TRACE_EXIT;
}

void StartZookeeperClient( void )
{
    const char method_name[] = "StartZookeeperClient";
    TRACE_ENTRY;

    int rc = -1;

    if ( ZClientEnabled )
    {
        if ( ZClient )
        {
            rc = ZClient->StartWork();
            if (rc == 0)
            {
                ZClient->StartMonitoring();

                char buf[MON_STRING_BUF_SIZE];
                snprintf(buf, sizeof(buf),
                         "[%s], ZClient node monitoring started\n"
                        , method_name);
                mon_log_write(MON_MONITOR_STARTZCLIENT_1, SQ_LOG_INFO, buf);
            }
        }
    }

    TRACE_EXIT;
}

#ifdef USE_SEQUENCE_NUM
long long CMonitor::GetTimeSeqNum()
{
    const char method_name[] = "CMonitor::GetTimeSeqNum";
    TRACE_ENTRY;

    TimeSeqNumLock_.lock();

    struct timespec currTime;
    clock_gettime(CLOCK_REALTIME, &currTime);

    if ( (currTime.tv_sec > savedTime_.tv_sec) ||
         ( (currTime.tv_sec == savedTime_.tv_sec) &&
           (currTime.tv_nsec > savedTime_.tv_nsec) ) )
    {
        savedTime_.tv_sec = currTime.tv_sec;
        savedTime_.tv_nsec = currTime.tv_nsec;
    }
    else
    {   // time drifted back. just increment nanoseconds
        savedTime_.tv_nsec++;
        // overflow check is not required as it would take
        // at least 3 billion requests to overflow
    }

    long long result;
    int* value = (int *)&result;
    value[0] = savedTime_.tv_nsec;  // little endian
    value[1] = savedTime_.tv_sec;

    TimeSeqNumLock_.unlock();

    if (trace_settings & TRACE_REQUEST_DETAIL)
        trace_printf("%s@%d Time seq num = %Lx\n", method_name, __LINE__, result);

    TRACE_EXIT;

    return result;
}
#endif

int main (int argc, char *argv[])
{
    int i;
    int rc;
    bool done = false;
    char *env;
    char *nodename = NULL;
    char fname[MAX_PROCESS_PATH];
    char short_node_name[MPI_MAX_PROCESSOR_NAME];
#ifndef NAMESERVER_PROCESS
    char port_fname[MAX_PROCESS_PATH];
    char temp_fname[MAX_PROCESS_PATH];
#endif
    char buf[MON_STRING_BUF_SIZE];
#ifndef NAMESERVER_PROCESS
    unsigned int initSleepTime = 1; // 1 second
#endif

    mallopt(M_ARENA_MAX, 4); // call to limit the number of arena's of  monitor to 4.This call doesn't seem to have any effect !

#ifdef NAMESERVER_PROCESS
    CALL_COMP_DOVERS(trafns, argc, argv);
#else
    CALL_COMP_DOVERS(monitor, argc, argv);
#endif

    const char method_name[] = "main";

#ifdef NAMESERVER_PROCESS
    IsNameServer = true;
#endif

    if (argc < 2) {
      printf("error: monitor needs an argument...exitting...\n");
      exit(0);
    }

    int lv_arg_index = 1;
    while ( lv_arg_index < argc )
    {
        // In installations like Cloudera Manager, the monitor is started in AGENT mode
        if ( strcmp( argv[lv_arg_index], "COLD_AGENT" ) == 0 )
        {
            IsAgentMode = true;
        }

        lv_arg_index++;
    }

    // Set flag to indicate whether we are operating in a real cluster
    // or a virtual cluster.   This is used throughout the monitor when
    // behavior differs for a real vs. virtual cluster environment.
    if ( !IsAgentMode )
    {
        if ( getenv( "SQ_VIRTUAL_NODES" ) )
        {
            IsRealCluster = false;
            Emulate_Down = true;
        }
        if ( IsRealCluster )
        {
            // The monitor processes may be started by MPIrun utility
            env = getenv("SQ_MON_CREATOR");
            if ( env != NULL && strcmp(env, "MPIRUN") == 0 )
            {
                IsMPIChild = true;
            }
            // The monitor can be set to run in AGENT mode
            env = getenv("SQ_MON_RUN_MODE");
            if ( env != NULL && strcmp(env, "AGENT") == 0 )
            {
                IsAgentMode = true;
            }
        }
#ifdef NAMESERVER_PROCESS
        else
        {
            env = getenv("SQ_MON_RUN_MODE");
            if ( env != NULL && strcmp(env, "AGENT") == 0 )
            {
                IsAgentMode = true;
            }
        }
#endif
    }

#ifndef NAMESERVER_PROCESS
    env = getenv("SQ_NAMESERVER_ENABLED");
    if ( env && isdigit(*env) )
    {
        NameServerEnabled = atoi(env);
    }
#endif

    if ( IsAgentMode || IsNameServer )
    {
        MON_Props xprops( true );
#ifdef NAMESERVER_PROCESS
        xprops.load( "nameserver.env" );
#else
        xprops.load( "monitor.env" );
#endif
        MON_Smap_Enum xenum( &xprops );
        while ( xenum.more( ) )
        {
            char *xkey = xenum.next( );
            const char *xvalue = xprops.get( xkey );
            if ( xkey && xkey[0] && xvalue )
            {
                setenv( xkey, xvalue, 1 );
            }
        }
    }

#ifdef NAMESERVER_PROCESS
    MonLog = new CMonLog( "log4cxx.monitor.trafns.config", "NS", "alt.mon", -1, -1, getpid(), "$TNS" );
#else
    MonLog = new CMonLog( "log4cxx.monitor.mon.config", "MON", "alt.mon", -1, -1, getpid(), "$MONITOR" );
#endif

    MonLog->setupInMemoryLog();

#ifdef MULTI_TRACE_FILES
    initVariableKey();
#endif

#ifdef DMALLOC
    util_dmalloc_start();
#endif

    // Save our execution path
    env = getenv("PWD");
    if ( env )
    {
        STRCPY(MyPath,env);
    }
    for (i = strlen(argv[0])-1; i >= 0; i--)
    {
        if (argv[0][i] == '/')
        {
            argv[0][i] = '\0';
            strncpy(MyPath,argv[0],sizeof(MyPath));
            argv[0][i] = '/';
            break;
        }
    }
    env = getenv("SQ_MEASURE");
    if ( env && *env == '1' )
    {
        Measure = 1;
        snprintf(fname, sizeof(fname), "%s/monitor.P%d",
                 getenv("MPI_TMPDIR"),getpid());
        setenv("MPI_INSTR", fname, 1);
    }
    if ( env && *env == '2' )
    {
        Measure = 2;
        snprintf(fname, sizeof(fname), "%s/monitor.cpu.P%d:cpu",
                 getenv("MPI_TMPDIR"),getpid());
        setenv("MPI_INSTR", fname, 1);
    }

    env = getenv("SQ_IC");
    if ( env != NULL && strcmp(env, "IBV") == 0 )
    {
        CommType = CommType_InfiniBand;
    }
    else
    {
        CommType = CommType_Sockets;
    }

    // Mask all allowed signals
    sigset_t              mask;
    sigfillset(&mask);
    sigdelset(&mask, SIGPROF); // allows profiling such as google profiler
    sigdelset(&mask, SIGUSR2);
    rc = pthread_sigmask(SIG_SETMASK, &mask, NULL);
    if (rc != 0)
    {
        char buf[MON_STRING_BUF_SIZE];
        snprintf(buf, sizeof(buf), "[%s], pthread_sigmask error=%d\n",
                 method_name, rc);
        mon_log_write(MON_MONITOR_MAIN_1, SQ_LOG_ERR, buf);
    }

    // Setup HP_MPI software license
    int key = 413675219; //413675218 to display banner
    MPI_Initialized(&key);

    // Initialize MPI environment
    MPI_Init (&argc, &argv);

#ifdef NAMESERVER_PROCESS
    if ( argc > 1 )
    {
        if ( strcmp(argv[1], "SQMON1.1") == 0 )
        {
            MyPNID = atoi( argv[2] );
            int arg = 1;
            for ( ; argv[arg+11]; arg++ )
            {
                argv[arg] = argv[arg+11];
            }
            argv[arg] = NULL;
            argc -= 11;
        }
    }
#else
    monitorArgc = argc;
    STRCPY(monitorArgv[0], "trafns");
    for ( int arg = 1; arg < argc; arg++ )
        STRCPY(monitorArgv[arg], argv[arg]);
#endif

    env = getenv("MON_PROF_ENABLE");
    if ( env )
    {
        mon_profiler_init(env); // LCOV_EXCL_LINE
    }

    env = getenv("MON_SNMP_ENABLE");
    if ( env )
    {
        genSnmpTrapEnabled = true;
    }

#ifndef NAMESERVER_PROCESS
    env = getenv("MON_INIT_SLEEP");
    if ( env && isdigit(*env) )
    {
        initSleepTime = atoi(env);
    }
#endif

    env = getenv("SQ_COLD_STANDBY_SPARE");
    if ( env && isdigit(*env) )
    {
        if ( strcmp(env,"0")==0 )
        {
            SpareNodeColdStandby = false;
        }
    }

#ifndef NAMESERVER_PROCESS
    // We need to delay some to make sure all monitor processes have initialized before
    // any monitor tries to perform an Allgather operation.
    sleep( initSleepTime );
#endif

    MPI_Comm_set_errhandler(MPI_COMM_WORLD, MPI_ERRORS_RETURN);
    MPI_Comm_set_errhandler(MPI_COMM_SELF, MPI_ERRORS_RETURN);
#ifndef NAMESERVER_PROCESS
    MPI_Comm_rank (MPI_COMM_WORLD, &MyPNID);
#endif

    MonLog->setPNid( MyPNID );

    gethostname(Node_name, MPI_MAX_PROCESSOR_NAME);
    char *tmpptr = Node_name;
    while ( *tmpptr )
    {
        *tmpptr = (char)tolower( *tmpptr );
        tmpptr++;
    }

    // Remove the domain portion of the name if any
    char str1[MPI_MAX_PROCESSOR_NAME];
    memset( str1, 0, MPI_MAX_PROCESSOR_NAME );
    memset( short_node_name, 0, MPI_MAX_PROCESSOR_NAME );
    strcpy (str1, Node_name );

    char *str1_dot = strchr( (char *) str1, '.' );
    if ( str1_dot )
    {
        memcpy( short_node_name, str1, str1_dot - str1 );
    }
    else
    {
        strcpy (short_node_name, str1 );
    }

    if (IsRealCluster) 
    {
      strcpy (Node_name, short_node_name );
    }

#ifdef MULTI_TRACE_FILES
    setThreadVariable( (char *)"mainThread" );
#endif

#ifdef NAMESERVER_PROCESS
    // Without mpi daemon the monitor has no default standard output.
    // We create a standard output file here.
    if ( IsRealCluster )
    {
        snprintf(fname, sizeof(fname), "%s/logs/trafns.%s.log",
                 getenv("TRAF_HOME"), Node_name);
    }
    else
    {
        snprintf(fname, sizeof(fname), "%s/logs/trafns.%d.%s.log",
                 getenv("TRAF_HOME"), MyPNID, Node_name);
    }
#else
    // Without mpi daemon the monitor has no default standard output.
    // We create a standard output file here.
    if ( IsRealCluster )
    {
        snprintf(fname, sizeof(fname), "%s/logs/sqmon.%s.log",
                 getenv("TRAF_HOME"), Node_name);
    }
    else
    {
        snprintf(fname, sizeof(fname), "%s/logs/sqmon.%d.%s.log",
                 getenv("TRAF_HOME"), MyPNID, Node_name);
    }
#endif
    remove(fname);
    if( freopen (fname, "w", stdout) == NULL )
    {
        char buf[MON_STRING_BUF_SIZE];
        snprintf(buf, sizeof(buf), "[%s], can't open stdout (%s).\n",
                 method_name, fname);
        mon_log_write(MON_MONITOR_MAIN_2, SQ_LOG_ERR, buf);
    }
    setlinebuf(stdout);

    // Send stderr output to same file as stdout.  (Note: the monitor does
    // not write to stderr but perhaps there could be components included in
    // the monitor build that do write to stderr.)
    if ( dup2 ( 1, 2 ) == -1 )
    {
        printf ( "dup2 failed for stderr: %s (%d)\n", strerror(errno), errno);
    }

    switch( CommType )
    {
        case CommType_Sockets: // Valid communication protocol
            break;
        case CommType_InfiniBand: // Currenly disabled - requires HA MPI
            //MPI_Open_port (MPI_INFO_NULL, MyMPICommPort);
        default:
            printf( "SQ_IC contains invalid communication protocol: %s\n"
                   , CommTypeString(CommType));
            abort();
    }

    if ((!IsAgentMode) && (argc > 3 && strcmp (argv[2], "-integrate") == 0))
    {
        switch( CommType )
        {
            case CommType_InfiniBand:
                if (argc == 4 && strstr(argv[3], "$port#"))
                {
                    SMSIntegrating = IAmIntegrating = true;
                    strcpy( IntegratingMonitorPort, argv[3] );
                }
                else
                {
                    printf ( "Invalid integrating monitor MPI port: %s\n", argv[3]);
                    abort();
                }
                break;
            case CommType_Sockets:
                if ( IsAgentMode || isdigit (*argv[3]) )
                {
                    // In agent mode and when re-integrating (node up), all
                    // monitors processes start as a cluster of 1 and join to the
                    // creator monitor to establish the real cluster.
                    // Therefore, MyPNID will always be zero then it is
                    // necessary to use the node name to obtain the correct
                    // <pnid> from the configuration which occurs when creating the
                    // CMonitor object down below. By setting MyPNID to -1, when the
                    // CCluster::InitializeConfigCluster() invoked during the creation
                    // of the CMonitor object it will set MyPNID using Node_name.
#ifdef NAMESERVER_PROCESS
                    if ( IsRealCluster )
                        MyPNID = -1;
#else
                    MyPNID = -1;
#endif
                    SMSIntegrating = IAmIntegrating = true;
                    strcpy( IntegratingMonitorPort, argv[3] );
                }
                else
                {
                    printf ( "Invalid integrating monitor socket port: %s\n", argv[3]);
                    abort();
                }
                break;
            default:
                // Programmer bonehead!
                abort();
        }
        if ( isdigit (*argv[4]) )
        {

            CreatorShellPid = atoi( argv[4] );
        }
        else
        {
            printf ( "Invalid creator shell pid: %s\n", argv[4]);
            abort();
        }
        if ( isdigit (*argv[5]) )
        {

            CreatorShellVerifier = atoi( argv[5] );
        }
        else
        {
            printf ( "Invalid creator shell verifier: %s\n", argv[5]);
            abort();
        }

        // Trace cannot be specified on startup command but need to
        // check for trace environment variable settings.
        MonTrace->mon_trace_init("0", NULL);

    }

    if (IsAgentMode)
    {
        if ( IsRealCluster )
        {
            MyPNID = -1;
        }
        CreatorShellPid = 1000; // per monitor.sh
        CreatorShellVerifier = 0;
    }

    if (argc == 3 && isdigit(*argv[2]) )
    {
        MonTrace->mon_trace_init(argv[2], "STDOUT");
    }
    else if (argc >= 3 && strcmp (argv[2], "TRACEFILE") == 0)
    {
        if (argc == 4 && isdigit (*argv[3]))
        {
            MonTrace->mon_trace_init(argv[3], NULL);
        }
    }
    else
    {   // Trace not specified on startup command but need to check
        // for trace environment variable settings.
        MonTrace->mon_trace_init("0", NULL);
    }

    if ((env = getenv("MON_SYNC_DELAY")) != NULL)
    {   // Sync cycle delay value specified
        int delay;
        errno = 0;
        delay = strtol(env, NULL, 10);
        if ( errno == 0) next_test_delay = delay;
    }
    if (trace_settings & TRACE_INIT)
        trace_printf("%s@%d next_test_delay=%ld\n", method_name, __LINE__,
                     next_test_delay);

#ifdef USE_SONAR
    if ( (env = getenv("SQ_SONAR") ) != NULL )
    {
        sonar_state_init();
        if ( IsRealCluster && sonar_verify_state(SONAR_ENABLED | SONAR_MONITOR_ENABLED))
        {  // Not a virtual cluster and sonar is enabled.

            if (trace_settings & TRACE_INIT)
            {
                trace_printf("%s@%d Enabling Sonar\n", method_name, __LINE__);
            }

            MonStats = new CMonSonarStats();
        }
    }
#endif
    if ( MonStats == NULL )
    {
        if (trace_settings & TRACE_INIT)
        {
            trace_printf("%s@%d Using non-Sonar counters\n",
                         method_name, __LINE__);
        }

        MonStats = new CMonStats();
    }

    // Normal process termination uses SIGKILL by default.  Environment
    // variable PROC_TERM_SIG can be used to override this.  Valid values
    // are SIGKILL or SIGTERM (as either string or integer values).
    int procTermSig = SIGKILL;
    env = getenv("PROC_TERM_SIG");
    if  ( env )
    {
        if (strcasecmp(env, "SIGKILL") == 0)
        {
            procTermSig = SIGKILL;
        }
        else if (strcasecmp(env, "SIGTERM") == 0)
        {
            procTermSig = SIGTERM;
        }
        else
        {
            procTermSig = atoi( env );
            if ( !(procTermSig == SIGKILL || procTermSig == SIGTERM) )
            {
                procTermSig = SIGKILL;
            }
        }
    }

    if (trace_settings & TRACE_INIT)
    {
        trace_printf("%s@%d Using signal %d for normal processes "
                     "termination.\n", method_name, __LINE__, procTermSig);
    }

    // Record statistics (sonar counters): monitor is busy
    if (sonar_verify_state(SONAR_ENABLED | SONAR_MONITOR_ENABLED))
       MonStats->MonitorBusyIncr();

#ifdef NAMESERVER_PROCESS
    snprintf(buf, sizeof(buf),
                 "[CMonitor::main], %s, Started! CommType: %s\n"
                , CALL_COMP_GETVERS2(trafns), CommTypeString( CommType ));
#else
    snprintf(buf, sizeof(buf),
                 "[CMonitor::main], %s, Started! CommType: %s (%s%s%s%s)\n"
                , CALL_COMP_GETVERS2(monitor)
                , CommTypeString( CommType )
                , IsRealCluster?"RealCluster":"VirtualCluster"
                , IsAgentMode?"/AgentMode":""
                , IsMPIChild?"/MPIChild":""
                , NameServerEnabled?"/NameServerEnabled":"" );
#endif
    mon_log_write(MON_MONITOR_MAIN_3, SQ_LOG_INFO, buf);

#ifdef DMALLOC
    if (trace_settings & TRACE_INIT)
       trace_printf("%s@%d" "DMALLOC Option set" "\n", method_name, __LINE__);
#endif

#ifdef DEBUGGING
    if (trace_settings & TRACE_INIT)
       trace_printf("%s@%d" "DEBUGGING Option set" "\n", method_name, __LINE__);
#endif

    if ( Emulate_Down )
        if (trace_settings & TRACE_INIT)
           trace_printf("%s@%d" "EMULATE_DOWN Option set" "\n", method_name, __LINE__);

    {
#ifndef NAMESERVER_PROCESS
        // Create thread for monitoring redirected i/o.
        // This is also used for monitor logs, so start it early.
        Redirector.start();
#endif

        // Create global configuration now
        ClusterConfig = new CClusterConfig();
        if (ClusterConfig)
        {
            bool traceEnabled = (trace_settings & TRACE_TRAFCONFIG) ? true : false;
            if (ClusterConfig->Initialize( traceEnabled, MonTrace->getTraceFileName()))
            {
                if (!ClusterConfig->LoadConfig())
                {
                     char la_buf[MON_STRING_BUF_SIZE];
                     sprintf(la_buf, "[%s], Failed to load cluster configuration.\n", method_name);
                     mon_log_write(MON_MONITOR_MAIN_12, SQ_LOG_CRIT, la_buf);

                     abort();
                }
            }
            else
            {
                char la_buf[MON_STRING_BUF_SIZE];
                sprintf(la_buf, "[%s], Failed to open cluster configuration.\n", method_name);
                mon_log_write(MON_MONITOR_MAIN_13, SQ_LOG_CRIT, la_buf);

                abort();
            }
        }
       else
       {
           char la_buf[MON_STRING_BUF_SIZE];
           sprintf(la_buf, "[%s], Failed to allocate cluster configuration.\n", method_name);
           mon_log_write(MON_MONITOR_MAIN_14, SQ_LOG_CRIT, la_buf);

           abort();
        }

        //Moved creation of the below to later on
        //Nodes = new CNodeContainer (); 
        //Config = new CConfigContainer ();
        //Monitor = new CMonitor (procTermSig);
        

        // Set up zookeeper and determine the master
        if ( IsAgentMode || IsRealCluster )
        {
            // Zookeeper client is enabled only in a real cluster
            env = getenv("SQ_MON_ZCLIENT_ENABLED");

            if ( env )
            {
                if ( env && isdigit(*env) )
                {
                    if ( strcmp(env,"0")==0 )
                    {
                        ZClientEnabled = false;
                    }
                }
            }

            if ( ZClientEnabled )
            {
                 CreateZookeeperClient( );
            }
        }
        else
        {
            ZClientEnabled = false;
        }

        if (IsAgentMode)
        {
            if ((ZClientEnabled) && (ZClient != NULL))
            {
                // Do not wait, just see if one exists
                const char *masterMonitor = ZClient->WaitForAndReturnMaster(false);

                if (masterMonitor)
                {
                    strcpy (MasterMonitorName, masterMonitor);

                    if (trace_settings & TRACE_INIT)
                    {
                        trace_printf("%s@%d (MasterMonitor) IsAgentMode = TRUE, masterMonitor from ZK: %s, Node_name: %s\n"
                                     , method_name
                                     , __LINE__
                                     , MasterMonitorName
                                     , Node_name);
                    }

                    // unfortunately, we have to do this to see if we are the master before
                    // other things are set up.   This is how we must do that
                    if (strcmp(Node_name, masterMonitor) == 0)
                    {
                        IsMaster = true;
                    }
                    else
                    {
                        IsMaster = false;
                    }
                }
                else
                {
                    strcpy (MasterMonitorName, ClusterConfig->GetConfigMasterByName());

                    if (trace_settings & TRACE_INIT)
                    {
                        trace_printf("%s@%d (MasterMonitor) IsAgentMode = TRUE, ConfigMasterMonitor: %s, Node_name:%s \n"
                                     , method_name
                                     , __LINE__
                                     , MasterMonitorName
                                     , Node_name);
                    }

                    if (strcmp (Node_name,  ClusterConfig->GetConfigMasterByName()) == 0)
                    {
                        IsMaster = true;
                    }
                    else
                    {
                        IsMaster = false;
                    }
                }
             }
#ifdef NAMESERVER_PROCESS
             else
             {
                strcpy (MasterMonitorName, ClusterConfig->GetConfigMasterByName());

                if (trace_settings & TRACE_INIT)
                {
                    trace_printf("%s@%d (MasterMonitor) IsAgentMode = TRUE, ConfigMasterMonitor: %s, Node_name:%s \n"
                                 , method_name
                                 , __LINE__
                                 , MasterMonitorName
                                 , Node_name);
                }

                if ( IsRealCluster )
                {
                    if (strcmp (Node_name,  ClusterConfig->GetConfigMasterByName()) == 0)
                    {
                        IsMaster = true;
                    }
                    else
                    {
                        IsMaster = false;
                    }
                 }
                 else
                 {
                     IsMaster = ( MyPNID == 0 );
                 }
             }
#endif
         }

         if (IsAgentMode)
         {
            if (!IsMaster)
            {
#ifdef NAMESERVER_PROCESS
                if ( IsRealCluster )
                {
                    MyPNID = -1;
                }
#else
                MyPNID = -1;
#endif
                SMSIntegrating = IAmIntegrating = true;
#ifdef NAMESERVER_PROCESS
                char *monitorPort = getenv ("NS_COMM_PORT");
#else
                char *monitorPort = getenv ("MONITOR_COMM_PORT");
#endif
                if (monitorPort)
                {
#ifdef NAMESERVER_PROCESS
                    if ( IsRealCluster )
                    {
                        strcpy( IntegratingMonitorPort, MasterMonitorName);
                    }
                    else
                    {
                        char localHost[MAX_PROCESSOR_NAME];
                        gethostname( localHost, MAX_PROCESSOR_NAME );
                        strcpy( IntegratingMonitorPort, localHost);
                    }
#else
                    strcpy( IntegratingMonitorPort, MasterMonitorName);
#endif
                    strcat( IntegratingMonitorPort, ":");
                    strcat( IntegratingMonitorPort, monitorPort);
                }
                if (trace_settings & TRACE_INIT)
                {
                    trace_printf( "%s@%d (MasterMonitor) IsAgentMode = TRUE, I am NOT the master, "
                                  "MyPNID=%d, master port=%s\n"
                                , method_name, __LINE__
                                , MyPNID, IntegratingMonitorPort );
                }
            }
            else
            {
                if (trace_settings & TRACE_INIT)
                {
                    trace_printf( "%s@%d (MasterMonitor) IsAgentMode = TRUE, I am the master, MyPNID=%d\n"
                                , method_name, __LINE__, MyPNID );
                }
                IAmIntegrating = false;
            }
        }

        NameServerConfig = new CNameServerConfigContainer ();
        Nodes = new CNodeContainer ();
        Config = new CConfigContainer ();
#ifdef NAMESERVER_PROCESS
        Monitor = new CMonitor ();
#else
        Monitor = new CMonitor (procTermSig);
#endif
#ifndef NAMESERVER_PROCESS
        if (NameServerEnabled)
        {
            NameServer = new CNameServer ();
            PtpClient = new CPtpClient ();
        }
#endif

        if ( IsAgentMode )
        {
            if (trace_settings & TRACE_INIT)
            {
                trace_printf( "%s@%d MyPNID=%d\n"
                            , method_name, __LINE__, MyPNID );
            }
            MonLog->setPNid( MyPNID );
        }

        if (IsAgentMode)
        {
            int monitorLead = -1;
            CNode *myNode = Nodes->GetNode(MyPNID);
            const char *masterMonitor = NULL;
            if (myNode == NULL)
            {
                char la_buf[MON_STRING_BUF_SIZE];
                sprintf( la_buf
                       , "[%s], Failed to get my Node, MyPNID=%d\n"
                       , method_name, MyPNID );
                mon_log_write(MON_MONITOR_MAIN_15, SQ_LOG_CRIT, la_buf);

                abort();
            }

            if ((ZClientEnabled) && (ZClient != NULL))
            {
                CNode *masterNode = Nodes->GetNode(MasterMonitorName);
                if (!masterNode)
                {
                    if (trace_settings & TRACE_INIT)
                    {
                          trace_printf("%s@%d (MasterMonitor) IsMaster=%d, masterNode is NULL, with MasterMonitorName %s\n", method_name, __LINE__, IsMaster, MasterMonitorName);
                    }
                    char la_buf[MON_STRING_BUF_SIZE];
                    sprintf(la_buf, "[%s], Failed to get my Master Node.\n", method_name);
                    mon_log_write(MON_MONITOR_MAIN_16, SQ_LOG_CRIT, la_buf);

                    abort();
                }
                else
                {
                    if (trace_settings & TRACE_INIT)
                    {
                          trace_printf("%s@%d (MasterMonitor) IsMaster=%d, masterNode=%s\n", method_name, __LINE__, IsMaster, masterNode->GetName() );
                    }
                }
                monitorLead = masterNode->GetPNid();
                if (MyPNID == monitorLead)
                {
                     ZClient->WatchNodeMasterDelete (myNode->GetName() ); // just in case of stale info
                     ZClient->CreateMasterZNode ( myNode->GetName() );
                     strcpy (MasterMonitorName, myNode->GetName());
                     ZClient->WatchMasterNode( MasterMonitorName );
                     if (trace_settings & TRACE_INIT)
                     {
                         trace_printf("%s@%d (MasterMonitor) IsMaster=%d, set monitor lead to %d\n", method_name, __LINE__, IsMaster, MyPNID);
                     }
                 }
                 else
                 {
                     masterMonitor = ZClient->WaitForAndReturnMaster(true);
                     CNode *masterNode = NULL;
                     if (masterMonitor)
                     {
                         strcpy (MasterMonitorName, masterMonitor);
                         masterNode = Nodes->GetNode(MasterMonitorName);
                     }

                     if (masterNode)
                     {
                          if (trace_settings & TRACE_INIT)
                          {
                              trace_printf("%s@%d (MasterMonitor) IsMaster=%d, set monitor lead to %d\n", method_name, __LINE__, IsMaster, masterNode->GetPNid());
                          }
                          monitorLead = masterNode->GetPNid();
                          ZClient->WatchMasterNode( MasterMonitorName ); 
                     }
                     else
                     {
                          if (trace_settings & TRACE_INIT)
                          {
                              trace_printf("%s@%d (MasterMonitor) IsMaster=%d, masterNode is NULL, with MasterMonitorName %s\n", method_name, __LINE__, IsMaster, MasterMonitorName);
                          }
                          char la_buf[MON_STRING_BUF_SIZE];
                          sprintf(la_buf, "[%s], Failed to get my Master Node.\n", method_name);
                          mon_log_write(MON_MONITOR_MAIN_17, SQ_LOG_CRIT, la_buf);

                          abort();
                     }
                }
            }
#ifdef NAMESERVER_PROCESS
            else
            {
                if ( !IsRealCluster )
                {
                    monitorLead = 0;
                }
            }
#endif
            char    buf[MON_STRING_BUF_SIZE];
            snprintf( buf, sizeof(buf)
                           , "[%s], Master Monitor is on node %d\n"
                           , method_name, monitorLead);
            mon_log_write(MON_MONITOR_MAIN_18, SQ_LOG_INFO, buf);
        }
        if (!IAmIntegrating)
        {
            Config->Init ();
        }
#ifndef NAMESERVER_PROCESS
        Devices = new CDeviceContainer ();
        if ( !Devices->IsInitialized() )
        {
            if ( IAmIntegrating )
            {   // Problem unmounting devices, let creator monitor know then abort
                Monitor->ReIntegrate( CCluster::Reintegrate_Err13 );
            }
            else
            {
                MPI_Abort(MPI_COMM_SELF,99); // too early to call failsafe node down.
            }
        }
#endif
        nodename = new char [Monitor->GetConfigPNodesCount() * MPI_MAX_PROCESSOR_NAME];

        // Create health check thread
        HealthCheck.start();

        // Create thread to accept connections from other monitors
        CommAccept.start();
#ifdef NAMESERVER_PROCESS
        // Create thread to accept connections from other name servers
        CommAcceptMon.start();
        if (IsMaster)
        {
            CommAcceptMon.startAccepting();
        }
#else
        if (NameServerEnabled)
        {
            // Create thread to accept point-2-point connections from other monitors
            PtpCommAccept.start();
        }
#endif
#ifndef NAMESERVER_PROCESS
        // Open file used to record process start/end times
        Monitor->openProcessMap ();
#endif

#ifndef NAMESERVER_PROCESS

        // Always using localio now, no other option
        SQ_theLocalIOToClient = new SQ_LocalIOToClient( MyPNID );
        assert (SQ_theLocalIOToClient);

        #define BLOCK_SIZE  512
        char *ioBuffer = NULL;
        int fd;

        rc = posix_memalign( (void**)&ioBuffer, BLOCK_SIZE, BLOCK_SIZE);
        if ( rc == -1 )
        {
            int err = rc;
            char buf[MON_STRING_BUF_SIZE];
            snprintf(buf, sizeof(buf), "[%s], can't allocate aligned %d bytes "
                     "buffer , Error=%d(%s)\n",
                     method_name, BLOCK_SIZE, err, ErrorMsg(err));
            mon_log_write(MON_MONITOR_MAIN_4, SQ_LOG_CRIT, buf);

            MPI_Abort(MPI_COMM_SELF,99);
        }

        memset( (void *)ioBuffer, 0 , BLOCK_SIZE );
#endif

// start ok
#ifndef NAMESERVER_PROCESS
        if (IsRealCluster)
        {
            snprintf(port_fname, sizeof(port_fname), "%s/monitor.port.%s",
                     getenv("MPI_TMPDIR"), short_node_name );
        }
        else
        {
            // Write out our port number so other processes can attach.
            snprintf(port_fname, sizeof(port_fname), "%s/monitor.port.%d.%s",
                     getenv("MPI_TMPDIR"),MyPNID,Node_name);
        }
#endif

        // Change Node_name what we have in our configuration
        CNode *myNode = Nodes->GetNode(MyPNID);
        if (myNode)
        {
            strcpy (Node_name, myNode->GetName());
        }

#ifndef NAMESERVER_PROCESS
        // create with no caching, user read/write, group read/write, other read
        fd = open( port_fname
                   , O_RDWR | O_TRUNC | O_CREAT | O_DIRECT
                   , S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH );
        if ( fd != -1 )
        {
            snprintf( ioBuffer, BLOCK_SIZE, "%s", MyCommPort );
            rc = write( fd, ioBuffer, BLOCK_SIZE );
            if ( rc == -1 )
            {
                int err = errno;
                char buf[MON_STRING_BUF_SIZE];
                snprintf(buf, sizeof(buf), "[%s], can't write %d bytes to "
                         "port file (%s), Error=%d(%s)\n",
                         method_name, BLOCK_SIZE, port_fname, err,
                         ErrorMsg(err));
                mon_log_write(MON_MONITOR_MAIN_5, SQ_LOG_CRIT, buf);

                if ( IAmIntegrating )
                    // This monitor is reintegrating into cluster.  Inform
                    // creator monitor of error, then abort.
                    Monitor->ReIntegrate( CCluster::Reintegrate_Err10 );
                else
                    MPI_Abort(MPI_COMM_SELF,99);
            }
            close( fd );
            if (trace_settings & TRACE_INIT)
                trace_printf("%s@%d" " Port file created, pnid=%d, port=%s" "\n", method_name, __LINE__, MyPNID, MyCommPort );
        }
        else
        {
            char buf[MON_STRING_BUF_SIZE];
            snprintf(buf, sizeof(buf), "[%s], can't open port file (%s), "
                     "Error= %s\n", method_name, port_fname,
                     ErrorMsg(errno));
            mon_log_write(MON_MONITOR_MAIN_6, SQ_LOG_CRIT, buf);

            if ( IAmIntegrating )
                // This monitor is reintegrating into cluster.  Inform
                // creator monitor of error, then abort.
                Monitor->ReIntegrate( CCluster::Reintegrate_Err11 );
            else
                MPI_Abort(MPI_COMM_SELF,99);
        }
        free( ioBuffer );
        int ret = SQ_theLocalIOToClient->initWorker();
        if (ret)
        {
            if (trace_settings & TRACE_INIT)
               trace_printf("%s@%d" " Cannot start localio worker, aborting "  "%d" "\n", method_name, __LINE__, ret);
            delete SQ_theLocalIOToClient;
            SQ_theLocalIOToClient = NULL;
            if ( IAmIntegrating )
                // This monitor is reintegrating into cluster.  Inform
                // creator monitor of error, then abort.
                Monitor->ReIntegrate( CCluster::Reintegrate_Err12 );
            else
                assert (false);
        }

        if (trace_settings & TRACE_INIT)
            trace_printf("%s@%d" "started LocalIOToClient environment\n" "\n", method_name, __LINE__);
#endif

        if (trace_settings & TRACE_INIT)
        {
            int rc = getrlimit( RLIMIT_SIGPENDING, &Rl );
            if ( rc == 0 )
            {
                printf("%s@%d" " RLIMIT_SIGPENDING cur=%d, max=%d\n", method_name, __LINE__, (int)Rl.rlim_cur, (int)Rl.rlim_max);
            }
        }

        if ( IAmIntegrating )
        {
            // This monitor is integrating to (joining) an existing cluster
            Monitor->ReIntegrate( 0 );
            MyNode->SetPhase( Phase_Activating );

            if (trace_settings & (TRACE_INIT | TRACE_RECOVERY))
                trace_printf("%s@%d" " After UpdateCluster" "\n", method_name, __LINE__);
        }
        else
        {
            Monitor->EnterSyncCycle();
            done = Monitor->exchangeNodeData();
            Monitor->ExitSyncCycle();

            if (trace_settings & (TRACE_INIT | TRACE_RECOVERY))
                trace_printf("%s@%d" " After ImAlive " "\n", method_name, __LINE__);
        }
    }

    // Enable Zookeeper client in real cluster only and
    // after the integration phase on a node 'up'
    if ( IsRealCluster )
    {
        if ( ZClientEnabled )
        {
            StartZookeeperClient();
            // Set watch for master
            if (IsAgentMode)
            {
                ZClient->WatchMasterNode( MasterMonitorName );
            }
            if (trace_settings & (TRACE_INIT | TRACE_RECOVERY))
            {
                trace_printf( "%s@%d (MasterMonitor) set watch for MasterMonitorName %s\n", method_name, __LINE__, MasterMonitorName );
            }
        }
    }

#ifndef NAMESERVER_PROCESS
    // Initialize Seabed disconnect semaphore
    char *port;
    switch( CommType )
    {
        case CommType_InfiniBand:
            port = strstr(MyCommPort, "$port#");
            if (port) port += 5;
            break;
        case CommType_Sockets:
            port = strchr(MyCommPort, ':');
            break;
        default:
            // Programmer bonehead!
            abort();
    }

    if (port != NULL)
    {
        int myPortNum;
        int semMax = 50;
        unsigned int segKey;

        myPortNum = strtol(&port[1], NULL, 10);

        env = getenv("MS_DISC_SEM");
        if ( env )
        {
            errno = 0;
            semMax = strtol(env, NULL, 10);
            if ( errno != 0) semMax = 50;
        }

        segKey = RobSem::getSegKey(0x40000000, myPortNum, MyPNID);
        rc = RobSem::create_sem(segKey, IPC_CREAT + IPC_EXCL, sbDiscSem, semMax);

        if ( rc != 0 )
        {
            char buf[MON_STRING_BUF_SIZE];
            snprintf(buf, sizeof(buf),
                     "[%s], Failed to create seabed disconnect semaphore: %s "
                     "(%d)\n", method_name, strerror(errno), errno);
            mon_log_write(MON_MONITOR_MAIN_10, SQ_LOG_CRIT, buf);
            sbDiscSem = NULL;
        }
    }
#endif

    // Create request worker threads
    CReqWorker::startReqWorkers();

#ifndef NAMESERVER_PROCESS
    if ( ! IAmIntegrating )
    {
        Monitor->StartPrimitiveProcesses();
    }

    env = getenv( "SQ_USE_CPU_AFFINITY" );
    if ( env && strcmp( env, "1" ) == 0 )
    {   // Set flag to indicate that logical node CPU affinity is used for
        // processes.
        // (see CNode::SetAffinity)
        usingCpuAffinity = true;
    }

    env = getenv( "SQ_USE_TSE_CPU_AFFINITY" );
    if ( env && strcmp( env, "1" ) == 0 )
    {   // Set flag to indicate that CPU affinity is used for TSE processes.
        // (see CNode::SetAffinity)
        usingTseCpuAffinity = true;
    }
#endif

    nice(MON_BASE_NICE);

    Monitor->setMonInitComplete(true);

    struct timeval awakenedAt;
    struct timeval awakeTime;
    struct timeval now;
    gettimeofday(&awakenedAt, NULL);

    while (!done)
    {
        // Record statistics (sonar counters): monitor is NOT busy
        if (sonar_verify_state(SONAR_ENABLED | SONAR_MONITOR_ENABLED))
           MonStats->MonitorBusyDecr();

        // Sleep for a maximum of next_test_delay.  Reduce
        // next_test_delay by the amount of time since awoke for
        // current sync cycle.  If current sync cycle took a long
        // time don't sleep at all.
        gettimeofday(&now, NULL);
        if ( (now.tv_usec - awakenedAt.tv_usec )  < 0 )
        {
            awakeTime.tv_sec = now.tv_sec - awakenedAt.tv_sec - 1;
            awakeTime.tv_usec = 1000000 + now.tv_usec - awakenedAt.tv_usec;
        }
        else
        {
            awakeTime.tv_sec = now.tv_sec - awakenedAt.tv_sec;
            awakeTime.tv_usec = now.tv_usec - awakenedAt.tv_usec;
        }

        if ( awakeTime.tv_sec == 0
             && awakeTime.tv_usec < next_test_delay)
        {
            usleep( next_test_delay - awakeTime.tv_usec );
        }
        gettimeofday(&awakenedAt, NULL);

        // Record statistics (sonar counters): monitor is busy
        if (sonar_verify_state(SONAR_ENABLED | SONAR_MONITOR_ENABLED))
           MonStats->MonitorBusyIncr();

#ifndef NAMESERVER_PROCESS
        Monitor->EnterSyncCycle();
        if ( Monitor->TmSyncPending() )
        {
            Monitor->TmSync ();
        }
        Monitor->ExitSyncCycle();
#endif

#ifndef NAMESERVER_PROCESS
        if ( !Monitor->GetPendingSlaveTmSync() &&
             Monitor->GetTotalSlaveTmSyncCount() == 0 )
        {
#endif
            Monitor->EnterSyncCycle();
            done = Monitor->exchangeNodeData();
            Monitor->ExitSyncCycle();
#ifndef NAMESERVER_PROCESS
        }
#endif

        if (done)
            break;


#ifndef NAMESERVER_PROCESS
        // Check to see if 'ckillall' is executing and disable the watchdog
        if ( !SQ_theLocalIOToClient->isWDTEnabled() )
        {
           // HealthCheck.setState(MON_EXIT_WATCHDOG);
        }
#endif

    }

    if (trace_settings & TRACE_STATS)
    {   // Write malloc statistics info to stderr
        monMallocStats();
    }

    if ( ZClientEnabled )
    {
        ZClient->StopMonitoring();
        ZClient->ShutdownWork();
    }

#ifndef NAMESERVER_PROCESS
    Redirector.shutdownWork();
#endif

    // shut down health check thread before shutting down reqWorker thread.
    HealthCheck.shutdownWork();

    ReqQueue.stats();
    // Stop request worker threads
    CReqWorker::shutdownWork();
    Replicator.stats();

    Monitor->stats();

#ifndef NAMESERVER_PROCESS
    // Tell the LIO worker threads to exit
    SQ_theLocalIOToClient->shutdownWork();
    if (NameServerEnabled)
    {
        PtpCommAccept.shutdownWork();
    }
#endif

    CommAccept.shutdownWork();
#ifdef NAMESERVER_PROCESS
    CommAcceptMon.shutdownWork();
#endif

#ifndef NAMESERVER_PROCESS
    // Rename the monitor "port" file
    sprintf(temp_fname, "%s.bak", port_fname);
    remove(temp_fname);
    rename(port_fname, temp_fname);
#endif

    delete [] nodename;
#ifndef NAMESERVER_PROCESS
    delete Devices;
#endif
    delete Nodes;
    delete ZClient;
    delete Monitor;
    Monitor = NULL; // TRACE uses this
    delete Config;

#ifndef NAMESERVER_PROCESS
    if ( sbDiscSem != NULL )
    {
        RobSem::destroy_sem( sbDiscSem );
    }
#endif

    if ( CommType == CommType_InfiniBand )
    {
        MPI_Close_port( MyCommPort );
    }
#if 0
    // TODO: MPICH cannot handle a node down and subsequent shutdown
    //       MPI_Finalize() hangs so its currently disabled, but
    //       causes an abnormal termination in the monitor process at exit.
    if (trace_settings & TRACE_INIT)
       trace_printf("%s@%d" "- Calling MPI_Finalize()" "\n", method_name, __LINE__);
    MPI_Finalize ();
#endif
#ifndef NAMESERVER_PROCESS
    if (trace_settings & TRACE_STATS)
    {
      trace_printf("%s@%d" "- LIO Stats: shared_buffers_total="  "%d" "\n", method_name, __LINE__, SQ_theLocalIOToClient->getSharedBufferCount());
      trace_printf("%s@%d" "- LIO Stats: shared_buffers_acquired_max="  "%d" "\n", method_name, __LINE__, SQ_theLocalIOToClient->getAcquiredBufferCount());
      trace_printf("%s@%d" "- LIO Stats: shared_buffers_available_min="  "%d" "\n", method_name, __LINE__, SQ_theLocalIOToClient->getAvailableBufferCount());
      trace_printf("%s@%d" "- LIO Stats: shared_buffer_misses="  "%d" "\n", method_name, __LINE__, SQ_theLocalIOToClient->getMissedBufferCount());
      trace_printf("%s@%d" "- LIO Stats: max queued child death=%d\n",
                   method_name, __LINE__,
                   SQ_theLocalIOToClient->getMaxChildDeathCount());
      trace_printf("%s@%d" "- LIO Stats: almost-dead pids=%d\n",
                   method_name, __LINE__,
                   SQ_theLocalIOToClient->getAlmostDeadPids());
      trace_printf("%s@%d" "- LIO Stats: verifierMap="  "%d" "\n", method_name, __LINE__, SQ_theLocalIOToClient->getVerifierMapCount());
    }
    delete SQ_theLocalIOToClient;
#endif

    snprintf(buf, sizeof(buf), "[CMonitor::main], Shutdown normally.\n");
    mon_log_write(MON_MONITOR_MAIN_11, SQ_LOG_INFO, buf);

    if (trace_settings & TRACE_STATS)
    {
        if (sonar_verify_state(SONAR_ENABLED | SONAR_MONITOR_ENABLED))
           MonStats->displayStats();
    }

    // Record statistics (sonar counters): monitor is NOT busy
    if (sonar_verify_state(SONAR_ENABLED | SONAR_MONITOR_ENABLED))
       MonStats->MonitorBusyDecr();

    delete MonLog;

    return 0;
}

#ifdef DELAY_TP
void CMonitor::Delay_TP(char *tpName)
{
    char nodename[20];
    char keyname[MAX_KEY_NAME];
    char valueC[MAX_VALUE_SIZE];
    int value;

    if (Config)
    {
       snprintf(nodename, sizeof(nodename), "NODE%d",MyPNID);
       CConfigGroup *group = Config->GetGroup(nodename);
       if (group)
       {
          strcpy(keyname,"REQUESTDELAY_TP");
          CConfigKey *key = group->GetKey(keyname);
          if (key)
          {
            strcpy(valueC,key->Value);
            value = atoi(valueC);
            sleep(value);
          }
       }
    }
}
#endif