@@ -2789,7 +2789,7 @@ void tm_process_msg(BMS_SRE *pp_sre)
 {
     short                  lv_ret;
     char                   la_send_buffer[4096];
-    char                   la_recv_buffer[sizeof(Tm_Req_Msg_Type)];
+    char                   la_recv_buffer[pp_sre->sre_reqDataSize];
     char                  *la_recv_buffer_ddl = NULL;
     Tm_Broadcast_Req_Type *lp_br_req;
     Tm_Broadcast_Rsp_Type *lp_br_rsp; 
// @@@ 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 <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <sys/time.h>

// General Seaquest includes
#include "SCMVersHelp.h"

// seabed includes
#include "seabed/ms.h"
#include "seabed/pctl.h"
#include "seabed/pevents.h"
#include "seabed/trace.h"
#include "seabed/thread.h"
#include "common/sq_common.h"

// tm includes
#include "rm.h"
#include "tmaudit.h"
// #include "tmmap.h"
#include "tmregistry.h"
#include "tmlogging.h"
#include "tmrecov.h"
#include "tmshutdown.h"
#include "tmpool.h"
#include "tmstats.h"

#include "tmglobals.h"
#include "tmtimer.h"
#include "tmthreadeg.h"

#include "tminfo.h"
#include "tmrecov.h"
#include "tmtxbranches.h"

#include "hbasetm.h"

extern void tm_xarm_initialize();
extern void tm_process_msg_from_xarm(CTmTxMessage * pp_msg);
extern int HbaseTM_initialize (bool pp_tracing, bool pv_tm_stats, CTmTimer *pp_tmTimer, short pv_nid);
extern int HbaseTM_initiate_stall(int where);
extern HashMapArray* HbaseTM_process_request_regions_info();


// Version
DEFINE_EXTERN_COMP_DOVERS(tm)
// global data 
CTmTxBranches     gv_RMs;
TM_MAP            gv_sync_map;
int32             gv_system_tx_count = 0;         
TM_Info           gv_tm_info;
SB_Int64_Type     gv_wait_interval = TM_DEFAULT_WAIT_INTERVAL;

CTmThreadExample  *gp_tmExampleThread;


// ---------------------------------------------------------------
// misc helper routines
// ---------------------------------------------------------------
void tm_fill_perf_stats_buffer (Tm_Perf_Stats_Rsp_Type *pp_buffer)
{

    int64 lv_tx_count = gv_tm_info.tx_count();
    int64 lv_abort_count = gv_tm_info.abort_count();
    int64 lv_commit_count = gv_tm_info.commit_count();
    int64 lv_current_tx_count =  gv_tm_info.current_tx_count();
    int32 lv_tm_abort_count =  gv_tm_info.tm_initiated_aborts();
    int32 lv_hung_tx_count = gv_tm_info.tx_hung_count();

    TMTrace (2, ("tm_fill_perf_stats_buffer : tx count : " PFLL ", abort count " PFLL ", commit count " PFLL ", current tx count " PFLL ".\n", lv_tx_count, lv_abort_count, lv_commit_count, lv_current_tx_count));

    pp_buffer->iv_error = 0;
    pp_buffer->iv_tx_count = lv_tx_count;
    pp_buffer->iv_abort_count = lv_abort_count;
    pp_buffer->iv_commit_count = lv_commit_count;
    pp_buffer->iv_tm_initiated_aborts = lv_tm_abort_count;
    pp_buffer->iv_hung_tx_count = lv_hung_tx_count;
    pp_buffer->iv_outstanding_tx_count = lv_current_tx_count;
    pp_buffer->iv_oldest_transid_1 = 0;
    pp_buffer->iv_oldest_transid_2 = 0;
    pp_buffer->iv_oldest_transid_3 = 0;
    pp_buffer->iv_oldest_transid_4 = 0;
}

void tm_fill_sys_status_buffer (Tm_Sys_Status_Rsp_Type *pp_buffer)
{

    int32 lv_up = 0;
    int32 lv_down = 0;
    if(gv_tm_info.state() == TM_STATE_UP) {
        lv_up = 1;
    }
    else if(gv_tm_info.state() == TM_STATE_DOWN ||
            gv_tm_info.state() == TM_STATE_WAITING_RM_OPEN) {
        lv_down = 1;
    }

    int32 lv_recovering = 0;
    if(gv_tm_info.sys_recov_state() != TM_SYS_RECOV_STATE_END) {
        lv_recovering = 1;
    }
    int32 lv_totaltms = 1;
    int32 lv_leadtm = 0;
    if(gv_tm_info.lead_tm()) {
        lv_leadtm = 1;
    }
    
    int32 lv_activetxns = gv_tm_info.num_active_txs();
    // If we're still in recovery we need to add any transactions 
    // still queued to recover.
    if (gv_tm_info.ClusterRecov())
       lv_activetxns += gv_tm_info.ClusterRecov()->txnStateList()->size();

    TMTrace (2, ("tm_fill_sys_status_buffer : up %d, down %d, recovering %d, activetxns %d.\n", lv_up, lv_down, lv_recovering, lv_activetxns));

    pp_buffer->iv_status_system.iv_up = lv_up;
    pp_buffer->iv_status_system.iv_down = lv_down;
    pp_buffer->iv_status_system.iv_recovering = lv_recovering;
    pp_buffer->iv_status_system.iv_totaltms = lv_totaltms;
    pp_buffer->iv_status_system.iv_activetxns = lv_activetxns;
    pp_buffer->iv_status_system.iv_leadtm = lv_leadtm;
}

// ---------------------------------------------------------
// tm_initialize_rsp_hdr
// Purpose : Initialize the header field for a message response.
// Note this should only be used for broadcast syncs now.
// All TM Library responses are through the CTmTxMessage class.
// ---------------------------------------------------------
int tm_initialize_rsp_hdr(short rsp_type, Tm_Rsp_Msg_Type *pp_rsp)
{
   pp_rsp->iv_msg_hdr.dialect_type = DIALECT_TM_SQ;
   pp_rsp->iv_msg_hdr.rr_type.reply_type = (short) (rsp_type + 1);
   pp_rsp->iv_msg_hdr.version.reply_version = TM_SQ_MSG_VERSION_CURRENT;
   pp_rsp->iv_msg_hdr.miv_err.error = FEOK;
   return FEOK;
}

// ----------------------------------------------------------------
// tm_start_auditThread
// Purpose : Start the audit thread.  
// ----------------------------------------------------------------
void tm_start_auditThread()
{
   char lv_name[20];

   TMTrace(2, ("tm_start_auditThread ENTRY\n")); 

   // Instantiate timer object
   sprintf(lv_name, "auditTh");
   CTmAuditObj *lp_auditobj = new CTmAuditObj(auditThread_main, (const char *) &lv_name);
   if (lp_auditobj)
   {
      gv_tm_info.tmAuditObj(lp_auditobj);
      gv_tm_info.initialize_adp();
   }
   else
   {
      tm_log_event(DTM_TMTIMER_FAILED, SQ_LOG_CRIT, "DTM_TMTIMER_FAILED");
      TMTrace(1, ("tm_start_auditThread - Failed to instantiate audit object.\n"));
      abort();
   }

   TMTrace(2, ("tm_start_auditThread EXIT. Timer thread %s(%p) started.\n", 
         lv_name, (void *) lp_auditobj));
} //tm_start_auditThread

// ----------------------------------------------------------------
// tm_start_timerThread
// Purpose : Start the timer thread.  This is used whenever an
// internal timed event occurs.
// ----------------------------------------------------------------
void tm_start_timerThread()
{
   char lv_name[20];

   TMTrace(2, ("tm_start_timerThread ENTRY\n")); 

   // Instantiate timer object
   sprintf(lv_name, "timerTh");
   CTmTimer *lp_timer = new CTmTimer(timerThread_main, -1, (const char *) &lv_name,
                                     gv_tm_info.timerDefaultWaitTime());
   if (lp_timer)
   {
      gv_tm_info.tmTimer(lp_timer);
      gv_startTime = lp_timer->startTime();
      lp_timer->addControlpointEvent(gv_tm_info.cp_interval());
      if (gv_tm_info.lead_tm())
      {
         TMTrace(2, ("tm_start_timerThread lead DTM, adding timer events\n")); 
         lp_timer->addStatsEvent(gv_tm_info.stats_interval());
      }
     lp_timer->addRMRetryEvent(gv_tm_info.RMRetry_interval());
   }
   else
   {
      tm_log_event(DTM_TMTIMER_FAILED, SQ_LOG_CRIT, "DTM_TMTIMER_FAILED");
      TMTrace(1, ("tm_start_timerThread - Failed to instantiate timer object.\n"));
      abort();
   }

   TMTrace(2, ("tm_start_timerThread EXIT. Timer thread %s(%p) started.\n", 
         lv_name, (void *) lp_timer));
} //tm_start_timerThread


// ----------------------------------------------------------------
// tm_start_exampleThread
// Purpose : Start the example thread.
// ----------------------------------------------------------------
void tm_start_exampleThread()
{
   char lv_name[20];

   TMTrace(2, ("tm_start_exampleThread ENTRY\n")); 

   // Instantiate thread object
   sprintf(lv_name, "exampleTh");
   gp_tmExampleThread = new CTmThreadExample(exampleThread_main, -2, (const char *) &lv_name);
   if (!gp_tmExampleThread)
   {
      tm_log_event(DTM_TMTIMER_FAILED, SQ_LOG_CRIT, "DTM_TMTIMER_FAILED");
      TMTrace(1, ("tm_start_exampleThread - Failed to instantiate example thread object\n"));
      abort();
   }

   TMTrace(2, ("tm_start_exampleThread EXIT. Example thread %s(%p) started.\n", 
         lv_name, (void *) gp_tmExampleThread));
} //tm_start_exampleThread


// ---------------------------------------------------------
// tm_send_reply
// Purpose :  send a reply
// Note this should only be used for cp and shutdown
// replies.
// ---------------------------------------------------------
void tm_send_reply(int32 pv_msgid, Tm_Rsp_Msg_Type *pp_rsp)
{
    int lv_len = sizeof(Tm_Rsp_Msg_Type);
    TMTrace( 2, ("tm_send_reply : ENTRY. msgid(%d), reply code(%d), error(%d).\n", 
                    pv_msgid, pp_rsp->iv_msg_hdr.rr_type.reply_type, 
                    pp_rsp->iv_msg_hdr.miv_err.error));

    BMSG_REPLY_(pv_msgid,                // msgid
                NULL,                    // replyctrl
                0,                       // replyctrlsize
                (char *) pp_rsp,         // replydata
                lv_len,                  // replydatasize
                0,                       // errorclass
                NULL);                   // newphandle
}

// ---------------------------------------------------------
// tm_up_check
// Purpose : Check that DTM is up and, if not, reply 
// to the client with an error.
// We need to allow active transasctions to continue to 
// completion when shutting down and when transactions are
// disabled.
// Note that this function will delete pp_msg if there is
// an error, so it must not be used after the call if false
// is returned.
// pv_block    default=true, block client in TM_LIB by returning
//          FESERVICEDISABLED.
//          false - don't block client on reply.
// --------------------------------------------------------
const bool TX_UNBLOCKED=false,
           TX_BLOCKED=true;

bool tm_up_check(CTmTxMessage * pp_msg, bool pv_block=true) 
{
    bool lv_up = false;
    short lv_replyCode = FEOK;

    switch (gv_tm_info.state())
    {
    case TM_STATE_UP:
    case TM_STATE_SHUTTING_DOWN:
    case TM_STATE_TX_DISABLED:
    case TM_STATE_TX_DISABLED_SHUTDOWN_PHASE1:
    case TM_SYS_RECOV_STATE_END:
    case TM_STATE_DRAIN:
       lv_replyCode = FEOK;
       lv_up = true;
       break;
    case TM_STATE_QUIESCE:
       lv_replyCode = (pv_block)?FESERVICEDISABLED:FETMFNOTRUNNING;
       lv_up = false;
       break;
    default:
       lv_replyCode = FETMFNOTRUNNING;
       lv_up = false;
    }
    
    if (!lv_up)
    {
       TMTrace(1, ("tm_up_check EXIT replying %d, up=%d.\n", lv_replyCode, lv_up));
       pp_msg->reply(lv_replyCode);
       delete pp_msg;
    }
    return lv_up;
}

// ---------------------------------------------------------
// tm_notx_check
// Purpose : Check that transaction exists, if not, reply 
// to the client with an error.  
// Note that this function will delete pp_msg if there is
// an error, so it must not be used after the call if false
// is returned
// --------------------------------------------------------
bool tm_notx_check(TM_TX_Info *pp_tx, CTmTxMessage *pp_msg)
{
   if (pp_tx == NULL)
   {
      TMTrace(1, ("tm_notx_check - unable to complete, returning error FENOTRANSID\n"));
      pp_msg->reply(FENOTRANSID);
      delete pp_msg;
      return false;
   }
   return true;
}

// ------------------------------------------------------------
// Process request methods
// ------------------------------------------------------------

// --------------------------------------------------------------
// process_req_abort
// Purpose : process message of type TM_MSG_TYPE_ABORTTRANSACTION
// --------------------------------------------------------------- 
void tm_process_req_abort(CTmTxMessage * pp_msg) 
{
    TM_Txid_Internal * lp_transid = (TM_Txid_Internal *)
                                    &pp_msg->request()->u.iv_abort_trans.iv_transid;
    
    TMTrace(2, ("tm_process_req_abort, ID %d ENTRY\n", lp_transid->iv_seq_num));

    if (!tm_up_check(pp_msg))
        return;
 
    TM_TX_Info *lp_tx = (TM_TX_Info*) gv_tm_info.get_tx(lp_transid);

    if (!tm_notx_check(lp_tx, pp_msg))
       return;

    lp_tx->stats()->txnAbort()->start();

    if (!gv_tm_info.multithreaded()) {
       lp_tx->req_abort(pp_msg);
       gv_tm_info.cleanup(lp_tx);
       delete pp_msg;
    }
    else
       lp_tx->queueToTransaction(lp_transid, pp_msg);

    TMTrace(2, ("tm_process_req_abort EXIT\n"));
}


// ----------------------------------------------------------------
// tm_process_req_registerregion
// Purpose : process message of type TM_MSG_TYPE_REGISTERREGION
// ----------------------------------------------------------------
void tm_process_req_registerregion(CTmTxMessage * pp_msg)
{
    TM_Txid_Internal * lp_transid = (TM_Txid_Internal *)
                                    &pp_msg->request()->u.iv_register_region.iv_transid;
    TM_Transseq_Type * lp_startid = (TM_Transseq_Type *)
                                    &pp_msg->request()->u.iv_register_region.iv_startid;

    TMTrace(2, ("tm_process_req_registerregion ENTRY for Txn ID (%d,%d), startid %ld, msgid %d\n",
                lp_transid->iv_node, lp_transid->iv_seq_num, (long) lp_startid, pp_msg->msgid()));
    TMTrace(3, ("tm_process_req_registerregion for Txn ID (%d,%d), startid %ld, with region %s\n",
                lp_transid->iv_node, lp_transid->iv_seq_num, (long) lp_startid,
                pp_msg->request()->u.iv_register_region.ia_regioninfo2)); 

    TM_TX_Info *lp_tx = (TM_TX_Info*) gv_tm_info.get_tx(lp_transid);

    if (!tm_notx_check(lp_tx, pp_msg))
       return;

    // lp_tx->req_registerRegion(pp_msg);
    if (!gv_tm_info.multithreaded()) {
       lp_tx->req_registerRegion(pp_msg);
       delete pp_msg;
    }
    else{
       lp_tx->queueToTransaction(lp_transid, pp_msg);
       // Protect the reply as we may be trying to reply at the same time in the main thread.
       lp_tx->lock();
       if (lp_tx->transactionBusy() && pp_msg->replyPending())
          pp_msg->reply(FEOK);
       lp_tx->unlock();
    }

    TMTrace(2, ("tm_process_req_registerregion EXIT\n")); 

} // tm_process_req_registerregion

// -----------------------------------------------------------------
// tm_process_req_ddlrequest
// Purpose : process message of type TM_MSG_TYPE_DDLREQUEST
// -----------------------------------------------------------------
void tm_process_req_ddlrequest(CTmTxMessage * pp_msg)
{
    TM_Txid_Internal * lp_transid = (TM_Txid_Internal *)
                                    &pp_msg->request()->u.iv_ddl_request.iv_transid;

    TMTrace(2, ("tm_process_req_ddlrequest ENTRY for Txn ID (%d, %d) ", lp_transid->iv_node, lp_transid->iv_seq_num));

    TM_TX_Info *lp_tx = (TM_TX_Info*) gv_tm_info.get_tx(lp_transid);

    if (!gv_tm_info.multithreaded()) {
       lp_tx->req_ddloperation(pp_msg);
       pp_msg->reply(FEOK);
       delete pp_msg;
    }
    else {
       lp_tx->queueToTransaction(lp_transid, pp_msg);
    }

    TMTrace(2, ("tm_process_req_ddlrequest EXIT for Txn ID"));
}

//-----------------------------------------------------------------
// tm_process_req_requestregioninfo
// Purpose: process message of type TM_MSG_TYPE_REQUESTREGIONINFO
//-----------------------------------------------------------------
void tm_process_req_requestregioninfo(CTmTxMessage * pp_msg)
{
   int64          lv_size = 0;
   void         **lp_tx_list = gv_tm_info.get_all_txs (&lv_size);
   union
   {
       int64 lv_transid_int64;
       TM_Txid_legacy lv_transid;
   } u;

   char tname[2000];
   tname[299] = '\0';
/*
   char ername[50], rname[100], offline[20], regid[200], hostname[200], port[100];
   tname[299] = '\0', ername[49] = '\0', rname[99] = '\0', offline[19] = '\0';
   regid[199]= '\0', hostname[199]='\0', port[99]='\0';
*/

   TMTrace(2, ("tm_process_req_requestregioninfo ENTRY.\n"));
   HashMapArray* map = HbaseTM_process_request_regions_info();
   TMTrace(2, ("tm_process_req_requestregioninfo HashMapArray call has finished.\n"));

   pp_msg->response()->u.iv_hbaseregion_info.iv_count = 0;
   for (int lv_inx = 0; ((lv_inx < TM_MAX_LIST_TRANS) && (lv_inx < lv_size)); lv_inx++)
   {
        TM_TX_Info *lp_current_tx = (TM_TX_Info *)lp_tx_list[lv_inx];
        if (!lp_current_tx)
            continue;
        pp_msg->response()->u.iv_hbaseregion_info.iv_trans[lv_inx].iv_status = lp_current_tx->tx_state();
        u.lv_transid.iv_seq_num = lp_current_tx->seqnum();
        pp_msg->response()->u.iv_hbaseregion_info.iv_trans[lv_inx].iv_transid = u.lv_transid_int64;
        pp_msg->response()->u.iv_hbaseregion_info.iv_trans[lv_inx].iv_nid = lp_current_tx->node();
        pp_msg->response()->u.iv_hbaseregion_info.iv_trans[lv_inx].iv_seqnum = lp_current_tx->seqnum();

        char* res2 = map->getRegionInfo(lp_current_tx->legacyTransid());
        if(strlen(res2) == 0)
            continue;
        strncpy(tname, res2, sizeof(tname) -1);
        strncpy(pp_msg->response()->u.iv_hbaseregion_info.iv_trans[lv_inx].iv_tablename, tname, sizeof(tname)-1);
/*

        char* res3 = map->getEncodedRegionName(lv_inx);
        strncpy(ername, res3, sizeof(ername) -1);
        strncpy(pp_msg->response()->u.iv_hbaseregion_info.iv_trans[lv_inx].iv_enc_regionname, ername, sizeof(ername)-1);

        char* res4 = map->getRegionName(lv_inx);
        strncpy(rname, res4, sizeof(rname) -1);
        strncpy(pp_msg->response()->u.iv_hbaseregion_info.iv_trans[lv_inx].iv_regionname, res4, strlen(res4));

        char* res5 = map->getRegionOfflineStatus(lv_inx);
        strncpy(offline, res5, sizeof(offline) -1);
        strncpy(pp_msg->response()->u.iv_hbaseregion_info.iv_trans[lv_inx].iv_is_offline, offline, sizeof(offline)-1);

        char* res6 = map->getRegionId(lv_inx);
        strncpy(regid, res6, sizeof(regid) -1);
        strncpy(pp_msg->response()->u.iv_hbaseregion_info.iv_trans[lv_inx].iv_region_id, regid, sizeof(regid)-1);

        char* res7 = map->getHostName(lv_inx);
        strncpy(hostname, res7, sizeof(hostname) -1);
        strncpy(pp_msg->response()->u.iv_hbaseregion_info.iv_trans[lv_inx].iv_hostname, hostname, sizeof(hostname)-1);

        char* res8 = map->getPort(lv_inx);
        strncpy(port, res8, sizeof(port) -1);
        strncpy(pp_msg->response()->u.iv_hbaseregion_info.iv_trans[lv_inx].iv_port, port, sizeof(port)-1);
*/

        pp_msg->response()->u.iv_hbaseregion_info.iv_count++;
   }

   if (lp_tx_list)
      delete []lp_tx_list;
   pp_msg->reply(FEOK);
   delete pp_msg;
   delete map;

   TMTrace(2, ("tm_process_req_requestregioninfo EXIT\n"));
}


// ----------------------------------------------------------------
// tm_process_req_GetNextSeqNum
// Purpose : Retrieve the next transaction sequence number
// block.  This is used to implement local transactions
// in Trafodion.
// ----------------------------------------------------------------
void tm_process_req_GetNextSeqNum(CTmTxMessage * pp_msg)
{
    TMTrace(2, ("tm_process_req_GetNextSeqNum ENTRY.\n")); 
    
    gv_tm_info.lock();
    gv_tm_info.tm_new_seqNumBlock(pp_msg->request()->u.iv_GetNextSeqNum.iv_block_size,
                                  &pp_msg->response()->u.iv_GetNextSeqNum.iv_seqNumBlock_start,
                                  &pp_msg->response()->u.iv_GetNextSeqNum.iv_seqNumBlock_count);
    gv_tm_info.unlock();
    pp_msg->reply(FEOK);

    TMTrace(2, ("tm_process_req_GetNextSeqNum EXIT returning Next seqNum start %d, block size %d\n",
            pp_msg->response()->u.iv_GetNextSeqNum.iv_seqNumBlock_start, 
            pp_msg->response()->u.iv_GetNextSeqNum.iv_seqNumBlock_count)); 
    delete pp_msg;
} // tm_process_req_GetNextSeqNum


// ----------------------------------------------------------------
// tm_process_req_begin
// Purpose : process message of type TM_MSG_TYPE_BEGINTRANSACTION
// ----------------------------------------------------------------
void tm_process_req_begin(CTmTxMessage * pp_msg) 
{
    short lv_error = FEOK;
    TMTrace(2, ("tm_process_req_begin ENTRY\n")); 
 
    if ((gv_tm_info.state() != TM_STATE_UP) ||
        (gv_tm_info.sys_recov_state() != TM_SYS_RECOV_STATE_END))
    {
        switch (gv_tm_info.state())
        {
        case TM_STATE_TX_DISABLED:
        case TM_STATE_DRAIN:
            lv_error = FEBEGINTRDISABLED;
            break;
        case TM_STATE_QUIESCE:
            lv_error = FESERVICEDISABLED;
            break;
        default:
            lv_error = FETMFNOTRUNNING;
        }
        TMTrace(1, ("tm_process_req_begin returning error %d.\n", lv_error));
        pp_msg->reply(lv_error);
        delete pp_msg;
        return;
    }

    // Instantiate a new tx object.
    TM_TX_Info *lp_tx = (TM_TX_Info *) gv_tm_info.new_tx(pp_msg->request()->u.iv_begin_trans.iv_nid, 
                                                       pp_msg->request()->u.iv_begin_trans.iv_pid,
                                                       -1, -1,
                                                       (void* (*)(long int)) &TM_TX_Info::constructPoolElement);

    // An error indicates we are handling our maximum number of concurrent
    // transactions.
    if (lp_tx == NULL)
    {
       // Removing this event for now as we keep hitting it and it's just announcing that 
       //   we've reached the maximum transactions allowed per node.
       //tm_log_event(DTM_TX_MAX_EXCEEDED, SQ_LOG_WARNING, "DTM_TX_MAX_EXCEEDED", 
       //             -1, /*error_code*/ 
       //             -1, /*rmid*/
       //             gv_tm_info.nid(), /*dtmid*/ 
       //             -1, /*seq_num*/
       //             -1, /*msgid*/
       //             -1, /*xa_error*/
       //             gv_tm_info.transactionPool()->get_maxPoolSize(), /*pool_size*/
       //             gv_tm_info.transactionPool()->totalElements() /*pool_elems*/);
       TMTrace(1, ("tm_process_req_begin, FETOOMANYTRANSBEGINS\n"));
       pp_msg->reply(FETOOMANYTRANSBEGINS);
       delete pp_msg;
       return;
    }
    lp_tx->lock();

    lp_tx->setAbortTimeout(pp_msg->request()->u.iv_begin_trans.iv_abort_timeout);
    lp_tx->TT_flags(pp_msg->request()->u.iv_begin_trans.iv_transactiontype_bits);

    // Start statistics counters
    lp_tx->stats()->txnTotal()->start();
    lp_tx->stats()->txnBegin()->start();

    //M8 eliminate the association with the transaction as there is
    // nothing more to do now that we don't support xa_start

    lp_tx->req_begin(pp_msg);
    lp_tx->unlock();

    // Since we're not queuing requests, we can delete pp_req here itself.
    delete pp_msg;

    TMTrace(2, ("tm_process_req_begin, ID (%d,%d), creator (%d,%d) EXIT\n", 
            lp_tx->node(), lp_tx->seqnum(), lp_tx->ender_nid(), lp_tx->ender_pid()));
}

// --------------------------------------------------------------
// process_req_doomtx
// Purpose : process message of type TM_MSG_TYPE_DOOMTX
// DOOMTRANSACTION marks the transaction for rollback and
// replies immediately.  Then in the background it drives 
// rollback.
// or ABORTTRANSACTION.  
// --------------------------------------------------------------- 
void tm_process_req_doomtx(CTmTxMessage * pp_msg) 
{
    TM_Txid_Internal * lp_transid = (TM_Txid_Internal *)
                                    &pp_msg->request()->u.iv_abort_trans.iv_transid;
    
   TMTrace(2, ("tm_process_req_doomtx ID %d ENTRY\n", lp_transid->iv_seq_num));

   if (!tm_up_check(pp_msg))
       return;
 
   TM_TX_Info *lp_tx = (TM_TX_Info*) gv_tm_info.get_tx(lp_transid);

   if (!tm_notx_check(lp_tx, pp_msg))
      return; 

   int16 lv_error = lp_tx->doom_txn();
   pp_msg->reply(lv_error);

   delete pp_msg;

   TMTrace(2, ("tm_process_req_doomtx EXIT\n"));
} //process_req_doomtx

// --------------------------------------------------------------
// process_req_TSE_doomtx
// Purpose : process message of type TM_MSG_TYPE_TSE_DOOMTX
// This is different from an application doomtransaction because
// it drives an immediate rollback.  This is necessary because 
// the TSE might be dooming the transaction because we have hit
// an audit threshold and can't allow the transaction to continue.
// --------------------------------------------------------------- 
void tm_process_req_TSE_doomtx(CTmTxMessage * pp_msg) 
{
   TM_Txid_Internal * lp_transid = (TM_Txid_Internal *)
                                    &pp_msg->request()->u.iv_abort_trans.iv_transid;
    
   TMTrace(2, ("tm_process_req_TSE_doomtx, ID %d ENTRY\n", lp_transid->iv_seq_num));

   if (!tm_up_check(pp_msg, TX_UNBLOCKED))
       return;
 
   TM_TX_Info *lp_tx = (TM_TX_Info*) gv_tm_info.get_tx(lp_transid);

   if (!tm_notx_check(lp_tx, pp_msg))
      return;

   if (lp_tx->isAborting())
   {
      TMTrace(1, ("tm_process_req_TSE_doomtx, already doomed.\n"));
      pp_msg->reply(FEOK);
      delete pp_msg;
   }
   else
   {
      int16 lv_error = lp_tx->doom_txn();
      pp_msg->reply(lv_error);
      lp_tx->queueToTransaction(lp_transid, pp_msg);
   }

   TMTrace(2, ("tm_process_req_TSE_doomtx EXIT\n"));
}

// --------------------------------------------------------------
// process_req_wait_tmup
// Purpose : Wait until the TM is up, and only then reply. 
// This can be used by an application to wait for DTM to be ready
// to process transactions.
// --------------------------------------------------------------- 
void tm_process_req_wait_tmup(CTmTxMessage * pp_msg) 
{
   TMTrace(2, ("tm_process_req_wait_tmup ENTRY\n"));

   if ((gv_tm_info.state() == TM_STATE_UP) &&
      (gv_tm_info.sys_recov_state() == TM_SYS_RECOV_STATE_END))
   {
      TMTrace(3, ("tm_process_req_wait_tmup : TM up, replying immediately.\n"));

      pp_msg->reply(FEOK);
      delete pp_msg;
   }
   else
   {
     TMTrace(3, ("tm_process_req_wait_tmup : Adding caller msgid(%d) to TMUP_Wait list.\n",
                      pp_msg->msgid()));
      gv_tm_info.TMUP_wait_list()->push(pp_msg);
   }

   TMTrace(2, ("tm_process_req_wait_tmup EXIT\n"));
} //process_req_wait_tmup

// --------------------------------------------------------------
// Purpose : process message of type TM_MSG_TYPE_ENDRANSACTION
// ---------------------------------------------------------------
void tm_process_req_end(CTmTxMessage * pp_msg) 
{
    TM_Txid_Internal * lp_transid = (TM_Txid_Internal *)
                                    &pp_msg->request()->u.iv_end_trans.iv_transid;
    
   TMTrace(1, ("tm_process_req_end, ID %d ENTRY\n", lp_transid->iv_seq_num));

    if (!tm_up_check(pp_msg))
       return;
 
    TM_TX_Info *lp_tx = (TM_TX_Info*) gv_tm_info.get_tx(lp_transid);

    if (!tm_notx_check(lp_tx, pp_msg))
       return;

    lp_tx->stats()->txnCommit()->start();

    if (!gv_tm_info.multithreaded()) {
       lp_tx->req_end(pp_msg);
       lp_tx->req_forget(pp_msg);
       gv_tm_info.cleanup(lp_tx);
       delete pp_msg;
    }
    else
      lp_tx->queueToTransaction(lp_transid, pp_msg);

   TMTrace(2, ("tm_process_req_end, ID %d EXIT\n", lp_transid->iv_seq_num));
}

// ------------------------------------------------------------------
// tm_process_req_join_trans
// Purpose : process message of type TM_MSG_TYPE_JOINTRANSACTION
// ------------------------------------------------------------------
void tm_process_req_join_trans(CTmTxMessage * pp_msg)
{
    TM_Txid_Internal *lp_transid = (TM_Txid_Internal *)
                                   &pp_msg->request()->u.iv_join_trans.iv_transid;

    TMTrace(2, ("tm_process_req_join_trans, ID %d, ENTRY\n",
                     lp_transid->iv_seq_num));

    if (!tm_up_check(pp_msg, TX_UNBLOCKED))
        return;

    TM_TX_Info *lp_tx = (TM_TX_Info *)gv_tm_info.get_tx(lp_transid);

    if (!tm_notx_check(lp_tx, pp_msg))
        return;
   
    // Call join in-line in main thread
    lp_tx->req_join(pp_msg);
    // Since we don't queue join requests, we can delete pp_req here itself.
    delete pp_msg;

    TMTrace(2, ("tm_process_req_join_trans EXIT\n"));
}


// -----------------------------------------------------------------
// tm_process_req_list
// Purpose : Process a list transactions request.
// ----------------------------------------------------------------
void tm_process_req_list(CTmTxMessage *pp_msg)
{
   int64          lv_size = 0;
   void         **lp_tx_list = gv_tm_info.get_all_txs (&lv_size);
   union
   {
       int64 lv_transid_int64;
       TM_Txid_legacy lv_transid;
   } u;

   TMTrace(2, ("tm_process_req_list ENTRY.\n"));

   pp_msg->response()->u.iv_list_trans.iv_count = 0;
   for (int lv_inx = 0; ((lv_inx < TM_MAX_LIST_TRANS) && (lv_inx < lv_size)); lv_inx++)
   {
        TM_TX_Info *lp_current_tx = (TM_TX_Info *)lp_tx_list[lv_inx];
        if (!lp_current_tx)
            break;
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_status = lp_current_tx->tx_state(); 
        u.lv_transid.iv_seq_num = lp_current_tx->seqnum();
        u.lv_transid.iv_node = lp_current_tx->node();
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_transid = u.lv_transid_int64;
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_nid = lp_current_tx->node();
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_seqnum = lp_current_tx->seqnum();
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_tag = lp_current_tx->tag();
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_owner_nid = lp_current_tx->ender_nid();
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_owner_pid = lp_current_tx->ender_pid();
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_event_count = lp_current_tx->eventQ()->size();
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_pendingRequest_count = lp_current_tx->PendingRequestQ()->size();
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_num_active_partic = lp_current_tx->num_active_partic();
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_num_partic_RMs = lp_current_tx->get_TSEBranchesParticCount();
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_XARM_branch = false; //TODO
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_transactionBusy = lp_current_tx->transactionBusy();
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_mark_for_rollback = lp_current_tx->mark_for_rollback();
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_tm_aborted = (lp_current_tx->tm_aborted()|lp_current_tx->tse_aborted());
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_read_only = lp_current_tx->read_only();
        pp_msg->response()->u.iv_list_trans.iv_trans[lv_inx].iv_recovering = lp_current_tx->recovering();
        pp_msg->response()->u.iv_list_trans.iv_count++;
   }
    
   if (lp_tx_list)
      delete []lp_tx_list;
   pp_msg->reply(FEOK);
   delete pp_msg;
   TMTrace(2, ("tm_process_req_list EXIT.\n"));
} //tm_process_req_list

//----------------------------------------------------------------
// tm_process_req_status_all_transmgmt
// Purpose : Process status of all transactions of type
// TM_MSG_TYPE_STATUSALLTRANSMGT
// ----------------------------------------------------------------

void tm_process_req_status_all_transmgmt(CTmTxMessage *pp_msg)
{
   int64        lv_size = 0;
   void       **lp_tx_list = gv_tm_info.get_all_txs (&lv_size);
   union
   {
        int64 lv_transid_int64;
        TM_Txid_legacy lv_transid;
   } u;

   TMTrace(2, ("tm_process_req_status_all_transmgmt ENTRY.\n"));

   pp_msg->response()->u.iv_status_alltrans.iv_count = 0;
   for (int lv_inx = 0; ((lv_inx < TM_MAX_LIST_TRANS) && (lv_inx < lv_size)); lv_inx++)
   {
       TM_TX_Info *lp_current_tx = (TM_TX_Info *)lp_tx_list[lv_inx];
       if (!lp_current_tx)
             break;
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_status = lp_current_tx->tx_state();
       u.lv_transid.iv_seq_num = lp_current_tx->seqnum();
       u.lv_transid.iv_node = lp_current_tx->node();
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_transid = u.lv_transid_int64;
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_timestamp = lp_current_tx->timestamp();
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_nid = lp_current_tx->node();
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_seqnum = lp_current_tx->seqnum();
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_tag = lp_current_tx->tag();
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_owner_nid = lp_current_tx->ender_nid();
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_owner_pid = lp_current_tx->ender_pid();
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_event_count = lp_current_tx->eventQ()->size();
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_pendingRequest_count = lp_current_tx->PendingRequestQ()->size();
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_num_active_partic = lp_current_tx->num_active_partic();
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_num_partic_RMs = lp_current_tx->get_TSEBranchesParticCount();
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_num_unresolved_RMs = lp_current_tx->get_TSEBranchesUnresolvedCount();
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_XARM_branch = false; //TODO
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_transactionBusy = lp_current_tx->transactionBusy();
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_mark_for_rollback = lp_current_tx->mark_for_rollback();
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_tm_aborted = (lp_current_tx->tm_aborted()|lp_current_tx->tse_aborted());
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_read_only = lp_current_tx->read_only();
       pp_msg->response()->u.iv_status_alltrans.iv_trans[lv_inx].iv_recovering = lp_current_tx->recovering();
       pp_msg->response()->u.iv_status_alltrans.iv_count++;
   }
                                                                                                                  if (lp_tx_list)
        delete []lp_tx_list;
   pp_msg->reply(FEOK);
   delete pp_msg;
   TMTrace(2, ("tm_process_req_status_all_transmgmt EXIT.\n"));

}// tm_process_req_status_all_transmgmt 


// -----------------------------------------------------------------
// tm_process_req_tmstats
// Purpose : process message of type TM_MSG_TYPE_TMSTATS to
// list TM statistics.
// ----------------------------------------------------------------
void tm_process_req_tmstats(CTmTxMessage *pp_msg)
{
   TMTrace(2, ("tm_process_req_tmstats ENTRY.\n"));

   gv_tm_info.stats()->readStats(&pp_msg->response()->u.iv_tmstats.iv_stats);

   if (pp_msg->request()->u.iv_tmstats.iv_reset) {
      gv_tm_info.clearCounts();
      gv_tm_info.stats()->clearCounters();
   }

   pp_msg->reply(FEOK);
   delete pp_msg;
   TMTrace(2, ("tm_process_req_tmstats EXIT.\n"));
} //tm_process_req_tmstats

// -----------------------------------------------------------------
// tm_process_req_attachrm
// Purpose : process message of type TM_MSG_TYPE_ATTACHRM to
// return the status of this TM.
// ----------------------------------------------------------------
void tm_process_req_attachrm(CTmTxMessage *pp_msg)
{
    TMTrace(2, ("tm_process_req_attachrm ENTRY fpr %s.\n", pp_msg->request()->u.iv_attachrm.ia_rmname));
 
    gv_tm_info.addTimerEvent(pp_msg, 0 /*execute now*/);

    pp_msg->reply(FEOK);
    delete pp_msg;

    TMTrace(2, ("tm_process_req_attachrm EXIT.\n"));
}

// -----------------------------------------------------------------
// tm_process_req_statustm
// Purpose : process message of type TM_MSG_TYPE_STATUSTM to
// return the status of this TM.
// ----------------------------------------------------------------
void tm_process_req_statustm(CTmTxMessage *pp_msg)
{
   RM_Info_TSEBranch *lp_rm;
   TMTrace(2, ("tm_process_req_statustm ENTRY.\n"));

   pp_msg->response()->u.iv_statustm.iv_status.iv_node = gv_tm_info.nid();
   pp_msg->response()->u.iv_statustm.iv_status.iv_isLeadTM = gv_tm_info.lead_tm();
   pp_msg->response()->u.iv_statustm.iv_status.iv_state = gv_tm_info.state();
   pp_msg->response()->u.iv_statustm.iv_status.iv_sys_recovery_state = gv_tm_info.sys_recov_state();
   pp_msg->response()->u.iv_statustm.iv_status.iv_shutdown_level = gv_tm_info.shutdown_level();
   pp_msg->response()->u.iv_statustm.iv_status.iv_incarnation_num = gv_tm_info.incarnation_num();

   pp_msg->response()->u.iv_statustm.iv_status.iv_number_active_txns = gv_tm_info.num_active_txs();
   // Pick up any queued indoubt transactions
   if (gv_tm_info.ClusterRecov())
      pp_msg->response()->u.iv_statustm.iv_status.iv_number_active_txns += gv_tm_info.ClusterRecov()->txnStateList()->size();

   pp_msg->response()->u.iv_statustm.iv_status.iv_is_isolated = gv_tm_info.leadTM_isolated();

   if (gv_RMs.TSE()->return_highest_index_used() == 0) {
      lp_rm = gv_RMs.TSE()->return_slot_by_index(0);
      if(lp_rm->in_use()) {
         pp_msg->response()->u.iv_statustm.iv_status.iv_rm_count = 1;
         lp_rm->copyto(&pp_msg->response()->u.iv_statustm.iv_status.ia_rminfo[0]);
      }
      else {
         pp_msg->response()->u.iv_statustm.iv_status.iv_rm_count =  0;
      }
   }
   else {
      pp_msg->response()->u.iv_statustm.iv_status.iv_rm_count = gv_RMs.TSE()->return_highest_index_used() + 1;
      for (int i=0; i<=gv_RMs.TSE()->return_highest_index_used(); i++)
      {
         lp_rm = gv_RMs.TSE()->return_slot_by_index(i);
         lp_rm->copyto(&pp_msg->response()->u.iv_statustm.iv_status.ia_rminfo[i]);
      }
   }

   pp_msg->reply(FEOK);
   delete pp_msg;
   TMTrace(2, ("tm_process_req_statustm EXIT.\n"));
} //tm_process_req_statustm


// -----------------------------------------------------------------
// tm_process_req_status_transmgmt
// Purpose : process message of type TM_MSG_TYPE_STATUS_TRANSMGMT to
// return the status of the transaction.
// ----------------------------------------------------------------
void tm_process_req_status_transmgmt(CTmTxMessage *pp_msg)
{
    TMTrace(2, ("tm_process_req_status_transmgmt ENTRY.\n"));
    TM_Txid_Internal *lp_transid = (TM_Txid_Internal *)
                                   &pp_msg->request()->u.iv_status_transm.iv_transid;
    TM_Transid lv_transid(*lp_transid);                               

    //should already be sent to the correct TM
    TM_TX_Info *lp_tx = (TM_TX_Info *)gv_tm_info.get_tx(lp_transid);
    if(!lp_tx) {
        pp_msg->reply(FENOTRANSID);
        delete pp_msg;
        return;
    }

    TM_Transid lv_fulltransid(*(lp_tx->transid()));                               

    lv_fulltransid.set_external_data_type(&pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_transid);
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_status               = lp_tx->tx_state();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_nid                  = lp_tx->node();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_seqnum               = lp_tx->seqnum();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_incarnation_num      = lv_transid.get_incarnation_num();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_tx_flags             = lv_transid.get_tx_flags();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_tt_flags             = lp_tx->TT_flags();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_owner_nid            = lp_tx->ender_nid();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_owner_pid            = lp_tx->ender_pid();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_event_count          = lp_tx->eventQ()->size();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_pendingRequest_count = lp_tx->PendingRequestQ()->size();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_num_active_partic    = lp_tx->num_active_partic();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_num_partic_RMs       = lp_tx->get_TSEBranchesParticCount();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_XARM_branch          = false;
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_transactionBusy      = lp_tx->transactionBusy();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_mark_for_rollback    = lp_tx->mark_for_rollback();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_tm_aborted           = lp_tx->tm_aborted();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_abort_flags          = lp_tx->abort_flags();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_read_only            = lp_tx->read_only();
    pp_msg->response()->u.iv_status_transm.iv_status_trans.iv_recovering           = lp_tx->recovering();

    pp_msg->reply(FEOK);
    delete pp_msg;
    TMTrace(2, ("tm_process_req_status_transmgmt EXIT.\n"));
} //tm_process_req_status_transmgmt

// -----------------------------------------------------------------
// tm_process_req_status_gettransinfo
// Purpose : process message of type TM_MSG_TYPE_GETTRANSINFO to
// return the trans ID information
// ----------------------------------------------------------------
void tm_process_req_status_gettransinfo(CTmTxMessage *pp_msg)
{
   TMTrace(2, ("tm_process_req_status_gettransinfo ENTRY.\n"));
   TM_Txid_Internal *lp_transid = (TM_Txid_Internal *)
                                   &pp_msg->request()->u.iv_status_transm.iv_transid;
   TM_Transid lv_transid(*lp_transid);
   union
   {
      int64 lv_tt_flags_int64;
      TM_TT_Flags lv_tt_flags;
   } u;
   
   //should already be sent to the correct TM
   TM_TX_Info *lp_tx = (TM_TX_Info *)gv_tm_info.get_tx(lp_transid);
   if(!lp_tx) {
      pp_msg->reply(FENOTRANSID);
      delete pp_msg;
      return;
   }
  
   TM_Transid lv_fulltransid(*(lp_tx->transid()));
   pp_msg->response()->u.iv_gettransinfo.iv_seqnum          = lp_tx->seqnum();
   pp_msg->response()->u.iv_gettransinfo.iv_node             = lp_tx->node();
   pp_msg->response()->u.iv_gettransinfo.iv_incarnation_num  = lv_fulltransid.get_incarnation_num();
   pp_msg->response()->u.iv_gettransinfo.iv_tx_flags         = lv_fulltransid.get_tx_flags();
   u.lv_tt_flags_int64                                       = lp_tx->TT_flags();
   pp_msg->response()->u.iv_gettransinfo.iv_tt_flags         = u.lv_tt_flags;
   pp_msg->response()->u.iv_gettransinfo.iv_version          = lv_fulltransid.get_version();
   pp_msg->response()->u.iv_gettransinfo.iv_checksum         = lv_fulltransid.get_check_sum();
   pp_msg->response()->u.iv_gettransinfo.iv_timestamp        = lv_fulltransid.get_timestamp();
  
   pp_msg->reply(FEOK);
   delete pp_msg;
   TMTrace(2, ("tm_process_req_status_gettransinfo EXIT.\n"));
} //tm_process_req_gettransinfo


// -----------------------------------------------------------------
// tm_process_req_leadtm
// Purpose : process message of type TM_MSG_TYPE_LEADTM to
// return the current Lead TMs nid.
// ----------------------------------------------------------------
void tm_process_req_leadtm(CTmTxMessage *pp_msg)
{
   TMTrace(2, ("tm_process_req_leadtm ENTRY.\n"));

   pp_msg->response()->u.iv_leadtm.iv_node = gv_tm_info.lead_tm_nid();

   pp_msg->reply(FEOK);
   delete pp_msg;
   TMTrace(2, ("tm_process_req_leadtm EXIT.\n"));
} //tm_process_req_leadtm


// -----------------------------------------------------------------
// tm_process_req_enabletrans
// Purpose : process message of type TM_MSG_TYPE_ENABLETRANS to
// enable transaction processing in DTM.
// This can only be executed by the Lead TM.  Non-lead TMs will 
// return FEDEVDOWN.
// ----------------------------------------------------------------
void tm_process_req_enabletrans(CTmTxMessage *pp_msg)
{
   short lv_error = FEOK;
   TMTrace(2, ("tm_process_req_enabletrans ENTRY.\n"));

   if (!gv_tm_info.lead_tm())
      lv_error = FEDEVDOWN;
   else
       switch (gv_tm_info.state())
       {
       case TM_STATE_QUIESCE:
       case TM_STATE_DRAIN:
         lv_error = FEINVALOP;
         break;
       case TM_STATE_TX_DISABLED:
       default:
         // Queue the enabletransaction to the timer thread for execution.
         gv_tm_info.addTimerEvent(pp_msg, 0 /*execute now*/);
       }

   // Reply immediately and leave the enable to run in the background.
   pp_msg->reply(lv_error);
   delete pp_msg;
   TMTrace(2, ("tm_process_req_enabletrans EXIT.\n"));
} //tm_process_req_enabletrans


// -----------------------------------------------------------------
// tm_process_req_disabletrans
// Purpose : process message of type TM_MSG_TYPE_DISABLETRANS to
// disable transaction processing in DTM.
// This can only be executed by the Lead TM.  Non-lead TMs will 
// return FEDEVDOWN.
// ----------------------------------------------------------------
void tm_process_req_disabletrans(CTmTxMessage *pp_msg)
{
   short lv_error = FEOK;
   char  lv_levelStr[20],
        *lp_levelStr = (char *) &lv_levelStr;
   switch (pp_msg->request()->u.iv_disabletrans.iv_shutdown_level)
   {
   case TM_DISABLE_SHUTDOWN_IMMEDIATE:
       strcpy(lp_levelStr, "Immediate");
       break;
   case TM_DISABLE_SHUTDOWN_NORMAL:
       strcpy(lp_levelStr, "Normal");
       break;
   default:
       strcpy(lp_levelStr, "** Invalid **");
   }
   TMTrace(2, ("tm_process_req_disabletrans ENTRY, level %s.\n", lp_levelStr));

   if (!gv_tm_info.lead_tm())
      lv_error = FEDEVDOWN;
   else
      if (gv_tm_info.state() == TM_STATE_UP ||
          gv_tm_info.state() == TM_STATE_TX_DISABLED ||
          (gv_tm_info.state() == TM_STATE_TX_DISABLED_SHUTDOWN_PHASE1 &&
           pp_msg->request()->u.iv_disabletrans.iv_shutdown_level == TM_DISABLE_SHUTDOWN_IMMEDIATE) ||
          gv_tm_info.state() == TM_STATE_DRAIN)
      {
         // For disabletrans normal shutdown, only queue the disabletransaction if all TMs have recovered.
         if (pp_msg->request()->u.iv_disabletrans.iv_shutdown_level != TM_DISABLE_SHUTDOWN_NORMAL ||
             gv_tm_info.all_tms_recovered())
            gv_tm_info.addTimerEvent(pp_msg, 0 /*execute now*/);
         else
            lv_error = FERETRY;
      }
      else
         lv_error = FEINVALOP;

   // Reply immediately and leave the disable to run in the background.
   pp_msg->reply(lv_error);
   delete pp_msg;
   TMTrace(2, ("tm_process_req_disabletrans EXIT, replied with error %d.\n", lv_error));
} //tm_process_req_disabletrans


// -----------------------------------------------------------------
// tm_process_req_draintrans
// Purpose : process message of type TM_MSG_TYPE_DRAINTRANS to
// drain transaction processing in this TM.
// This can be executed in any TM.  It is used to allow transactions
// to complete before a planned node outage.
// Immediate means abort all active transactions and overrides a 
// prior drain
// ----------------------------------------------------------------
void tm_process_req_draintrans(CTmTxMessage *pp_msg)
{
   short lv_error = FEOK;
   TMTrace(2, ("tm_process_req_draintrans ENTRY immediate=%d.\n", 
       pp_msg->request()->u.iv_draintrans.iv_immediate));

   if (gv_tm_info.state() == TM_STATE_UP ||
       gv_tm_info.state() == TM_STATE_DRAIN)
      gv_tm_info.drainTrans(pp_msg);
   else
      lv_error = FEINVALOP;

   // Reply immediately and leave the drain to run in the background.
   pp_msg->reply(lv_error);
   delete pp_msg;
   TMTrace(2, ("tm_process_req_draintrans EXIT error=%d.\n", lv_error));
} //tm_process_req_draintrans


// -----------------------------------------------------------------
// tm_process_req_status
// Purpose : process message of type TM_MSG_TYPE_STATUSTRANSACTION
// ----------------------------------------------------------------
void tm_process_req_status(CTmTxMessage * pp_msg) 
{
    TM_Txid_Internal *lp_transid = (TM_Txid_Internal *)
                                   &pp_msg->request()->u.iv_status_trans.iv_transid;

   TMTrace(2, ("tm_process_req_status, ID %d, ENTRY\n", 
                     lp_transid->iv_seq_num));

    if (!tm_up_check(pp_msg))
       return;

    TM_TX_Info *lp_tx = (TM_TX_Info *)gv_tm_info.get_tx(lp_transid);

    pp_msg->response()->u.iv_status_trans.iv_status = TM_TX_STATE_NOTX;

    if (!tm_notx_check(lp_tx, pp_msg))
       return;

    // Handle status request in main thread to avoid status
    // getting queued behind other requests.
    lp_tx->req_status(pp_msg);
    // Since we don't queue status requests, we can delete pp_msg here itself.
    delete pp_msg;
    TMTrace(2, ("tm_process_req_status EXIT\n"));
}

// ---------------------------------------------------------------
// process_req_suspend_trans
// Purpose : process request of type TM_MSG_TYPE_SUSPENDTRANSACTION
// ---------------------------------------------------------------
void tm_process_req_suspend_trans (CTmTxMessage * pp_msg)
{
    TM_Txid_Internal *lp_transid = (TM_Txid_Internal *)
                                   &pp_msg->request()->u.iv_suspend_trans.iv_transid;

    TMTrace(2, ("tm_process_req_suspend_trans, ID %d, ENTRY\n",
                  lp_transid->iv_seq_num));

    if (!tm_up_check(pp_msg))
        return;
     
    TM_TX_Info *lp_tx = (TM_TX_Info *)gv_tm_info.get_tx(lp_transid);

    if (!tm_notx_check(lp_tx, pp_msg))
        return;

    // Call suspend in-line in main thread
    lp_tx->req_suspend(pp_msg);
    // Since we don't queue suspend requests, we can delete pp_msg here itself.
    delete pp_msg;

    TMTrace(2, ("tm_process_req_suspend_trans EXIT\n"));
}


// -----------------------------------------------------------------
// tm_process_req_broadcast
// Purpose - process a broadcast for sync data
// ----------------------------------------------------------------
void tm_process_req_broadcast (BMS_SRE *pp_sre,
         Tm_Broadcast_Req_Type *pp_req, Tm_Broadcast_Rsp_Type *pp_rsp)
{
    TMTrace(2, ("tm_process_req_broadcast for node %d ENTRY\n",
                   pp_req->iv_node));

    ushort lv_len = sizeof(Tm_Broadcast_Rsp_Type);

    gv_tm_info.unpack_sync_buffer (pp_req, pp_req->iv_node);
    if (pp_req->iv_state_up)  // last one, can be considered up
    {
        gv_tm_info.can_takeover(true);
        gv_tm_info.tm_up(); // up for processing
    }
    
    XMSG_REPLY_(pp_sre->sre_msgId,       // msgid
                NULL,                    // replyctrl
 
               0,                       // replyctrlsize
                (char *) pp_rsp,         // replydata
                lv_len,                  // replydatasize
                0,                       // errorclass
                NULL);                   // newphandle
    TMTrace(2, ("tm_process_req_broadcast EXIT\n")); 
}
// --------------------------------------------------------------------
// ax_* methods
// --------------------------------------------------------------------

// --------------------------------------------------------------------
// tm_process_req_ax_reg
// Purpose : process message of type TM_MSG_TYPE_AX_REG
// --------------------------------------------------------------------
void tm_process_req_ax_reg (CTmTxMessage * pp_msg) 
{
   short lv_error = FEOK;
   int lv_ptype = -1;
   int lv_nid = -1;
   int lv_pid = -1;
   int lv_seq_num = 0;
   int32 lv_rmid = pp_msg->request()->u.iv_ax_reg.iv_rmid;
   TM_Txid_Internal *lp_transid = (TM_Txid_Internal *)
                                 &pp_msg->request()->u.iv_ax_reg.iv_txid;

   TMTrace(2, ("tm_process_req_ax_reg, ID (%d,%d), ENTRY msgid %d\n",
                   lp_transid->iv_node, lp_transid->iv_seq_num, pp_msg->msgid()));

   // Removed check here because ax_reg needs to work during system recovery in M6.
   //if (!tm_up_check(pp_msg))
   //    return; 

   TM_TX_Info *lp_tx = (TM_TX_Info *)gv_tm_info.get_tx(lp_transid);

   // sent to the wrong TM or this tx never existed or has been forgotten.
   if (lp_tx == NULL)
   {
      pp_msg->response()->u.iv_ax_reg.iv_TM_incarnation_num = gv_tm_info.incarnation_num();
      pp_msg->response()->u.iv_ax_reg.iv_LeadTM_nid = gv_tm_info.lead_tm_nid();


      if (pp_msg->request()->u.iv_ax_reg.iv_flags & TM_TT_NO_UNDO)
         lv_error = FEWRONGID;
      else
         lv_error = FEINVTRANSID;

      TMTrace(3, ("tm_process_req_ax_reg, ID (%d,%d) from RM %d not found in transactionPool - "
              "redirecting TSE to Lead TM,  error %d.\n",
              lp_transid->iv_node, lp_transid->iv_seq_num, lv_rmid, lv_error));
      //tm_log_event(DTM_TM_NO_TRANS, SQ_LOG_WARNING, "DTM_TM_NO_TRANS",
      //             lv_error,lv_rmid,lp_transid->iv_node,lp_transid->iv_seq_num);

      pp_msg->reply(lv_error);
      delete pp_msg;
      return;
   }
   lp_tx->stats()->ax_reg()->start();

   // The TSE doesn't always know its rmid, so we can't rely on that.
   // Instead we lookup the RM in out list.
   if (lv_rmid == -1 || lv_rmid == 0)
   {
      lv_error = BMSG_GETREQINFO_(MSGINFO_PTYPE, pp_msg->msgid(), &lv_ptype);
      if (!lv_error && lv_ptype == MS_ProcessType_TSE)
      {
         lv_error = BMSG_GETREQINFO_(MSGINFO_NID, pp_msg->msgid(), &lv_nid);
         if (!lv_error)
            lv_error = BMSG_GETREQINFO_(MSGINFO_PID, pp_msg->msgid(), &lv_pid);
         if (lv_error)
         {
            TMTrace(1, ("tm_process_req_ax_reg, Error %d retrieving nid "
                    "(%d) and pid (%d) for TSE.  ax_reg ignored.\n",
                    lv_error, lv_nid, lv_pid));
            tm_log_event(DTM_AX_REG_NID_PID_BAD, SQ_LOG_CRIT, "DTM_AX_REG_NID_PID_BAD", 
                lv_error, lv_rmid, -1, -1, pp_msg->msgid(), -1, -1, -1, -1, -1, -1, 
                -1, -1, -1, lv_pid, -1, NULL, lv_nid);
            pp_msg->reply(FENOTFOUND);
            delete pp_msg;
            return;
         }

         lv_rmid = gv_RMs.TSE()->return_rmid(lv_nid, lv_pid);

         if (lv_rmid == -1)
          {
            TMTrace(1, ("tm_process_req_ax_reg, RM not found in RM list. "
                    "ax_reg ignored.\n"));
            tm_log_event(DTM_AX_REG_NOTFOUND, SQ_LOG_CRIT, "DTM_AX_REG_NOTFOUND", 
                -1, lv_rmid, -1, -1, pp_msg->msgid(), -1, -1, -1, -1, -1, -1, 
                -1, -1, -1, lv_pid, -1, NULL, lv_nid);
            pp_msg->reply(FENOTFOUND);
            delete pp_msg;
            return;
         }
         else  
            TMTrace(3, ("tm_process_req_ax_reg, TSE ax_reg for rmid %d, TSE (%d, %d).\n",
                    lv_rmid, lv_nid, lv_pid));
      }
      else // Not TSE or error
      {
         if (!lv_error)
            lv_error = BMSG_GETREQINFO_(MSGINFO_NID, pp_msg->msgid(), &lv_nid);
         if (!lv_error)
            lv_error = BMSG_GETREQINFO_(MSGINFO_PID, pp_msg->msgid(), &lv_pid);
         if (lv_error)
          {
            TMTrace(1, ("tm_process_req_ax_reg, Error %d retrieving PTYPE (%d), "
                    "nid (%d) or pid (%d). ax_reg ignored!\n",
                    lv_error, lv_ptype, lv_nid, lv_pid));
            tm_log_event(DTM_AX_REG_PTYPE_BAD, SQ_LOG_CRIT, "DTM_AX_REG_PTYPE_BAD", 
                lv_error,-1,-1,-1,pp_msg->msgid(),-1,-1,-1,-1,-1,-1,-1,-1,lv_pid,
                lv_ptype,-1,NULL,lv_nid);
            pp_msg->reply(FENOTFOUND);
            delete pp_msg;
            return;
         }
         else // Not an error - ax_reg from XARM library and should contain the rmid.
              // but not yet implemented!
         {
            TMTrace(1, ("tm_process_req_ax_reg, Received unexpected ax_reg from non-TSE"
                    " process (%d, %d), PTYPE %d assuming this was an XARM request!?, ignored!\n",
                    lv_nid, lv_pid, lv_ptype));
            tm_log_event(DTM_AX_REG_XARM_NOTSUPPORTED, SQ_LOG_CRIT, "DTM_AX_REG_XARM_NOTSUPPORTED", 
                -1,pp_msg->request()->u.iv_ax_reg.iv_rmid,-1,pp_msg->msgid(),
                -1,-1,-1,-1,-1,-1,-1,-1,lv_pid,lv_ptype,0,lv_nid);
            pp_msg->reply(FENOTFOUND);
            delete pp_msg;
            return;
  }
      }
   }

   // Save the rmid back in the message
   pp_msg->request()->u.iv_ax_reg.iv_rmid = lv_rmid;
   // Call directly in the main thread to improve performance.
   //lp_tx->queueToTransaction(lp_transid, pp_msg);
   lp_tx->req_ax_reg(pp_msg);
   lv_seq_num = lp_transid->iv_seq_num;
   delete pp_msg;
   lp_tx->stats()->ax_reg()->stop();


   TMTrace(2, ("tm_process_req_ax_reg, ID %d, EXIT\n", lv_seq_num));
} //tm_process_req_ax_reg


// --------------------------------------------------------------------
// tm_process_req_ax_unreg
// Purpose : process message of type TM_MSG_TYPE_AX_UNREG
// --------------------------------------------------------------------
void tm_process_req_ax_unreg (CTmTxMessage * pp_msg) 
{
 
  TMTrace(2, ("tm_process_req_ax_unreg ENTRY\n"));

  // sorry, not implemented right now! 
  pp_msg->reply(FEOK);
  delete pp_msg;

  TMTrace(2, ("tm_process_req_ax_unreg EXIT\n"));
}

// ------------------------------------------------------------------
// callback methods and processing downline from callbacks
// ------------------------------------------------------------------

// ------------------------------------------------------------------
// tm_sync_cb
// Purpose : this method is registered with seabed and is used when
//           a sync is received (Phase 1)
// ------------------------------------------------------------------
int32 tm_sync_cb (void *pp_data, int32 pv_len , int32 pv_handle)
{
    Tm_Sync_Header  *lp_hdr = (Tm_Sync_Header*)pp_data;
    Tm_Sync_Data    *lp_sync_data = new Tm_Sync_Data;
    Tm_Sync_Data    *lp_data = (Tm_Sync_Data *)pp_data;

    pv_len = pv_len; // intel compiler warning 869

    if (pp_data == NULL)
    {
        tm_log_event(DTM_SYNC_INVALID_DATA, SQ_LOG_CRIT, "DTM_SYNC_INVALID_DATA");
        TMTrace(1, ("tm_sync_cb : data is invalid\n"));
        abort ();
    }

    TMTrace(2, ("tm_sync_cb ENTRY : type %d\n", lp_hdr->iv_type));
       
    // allow duplicates per Charles
    Tm_Sync_Data *lp_existing_data = (Tm_Sync_Data *)gv_sync_map.get(pv_handle);
    if (lp_existing_data != NULL)
    {
        delete lp_sync_data;
        return 0;
    }
    
    switch (lp_hdr->iv_type)
    {
        case TM_BEGIN_SYNC:
        case TM_END_SYNC:
        case TM_FORGET_SYNC:
        {
#ifdef DEBUG_MODE
            bool lv_test = false;
            ms_getenv_bool("TM_TEST_SINGLE_FORCE_ABORT", &lv_test);
            if (lv_test)
            {
            //   sprintf(la_buf, "TM Test: Force Abort\n");
            //   tm_log_write(DTM_TM_TEST_FORCE_ABORT, SQ_LOG_CRIT, la_buf);
             //  TMTrace(1, ("tm_sync_cb - %s", la_buf));
               abort ();
            }
#endif
            if (lp_data->u.iv_tx_data.iv_pid <= 0)
            {
               tm_log_event (DTM_SYNC_INVALID_PID, SQ_LOG_CRIT, "DTM_SYNC_INVALID_PID",
                     -1, /*error_code*/ 
                    -1, /*rmid*/
                    -1, /*dtmid*/ 
                    -1, /*seq_num*/
                    -1, /*msgid*/
                    -1, /*xa_error*/
                    -1, /*pool_size*/
                    -1, /*pool_elems*/
                    -1, /*msg_retries*/
                    -1, /*pool_high*/
                    -1, /*pool_low*/
                    -1, /*pool_max*/
                    -1, /*tx_state*/
                    lp_data->u.iv_tx_data.iv_pid); /*data */

               TMTrace(1, ("tm_sync_cb - Invalid sync PID: %d\n", lp_data->u.iv_tx_data.iv_pid));
               abort ();
            }
            if (lp_data->u.iv_tx_data.iv_transid.id[0] <= 0)
            {
               tm_log_event (DTM_SYNC_INVALID_TRANSID, SQ_LOG_CRIT, "DTM_SYNC_INVALID_TRANSID",
                    -1, /*error_code*/ 
                    -1, /*rmid*/
                    -1, /*dtmid*/ 
                    -1, /*seq_num*/
                    -1, /*msgid*/
                    -1, /*xa_error*/
                    -1, /*pool_size*/
                    -1, /*pool_elems*/
                    -1, /*msg_retries*/
                    -1, /*pool_high*/
                    -1, /*pool_low*/
                    -1, /*pool_max*/
                    -1, /*tx_state*/
                    -1, /*data */
                    -1, /*data1*/
                    -lp_data->u.iv_tx_data.iv_transid.id[0]);/*data2 */
               TMTrace(1, ("tm_sync_cb - Invalid sync Trans ID: " PFLL "\n", 
                           lp_data->u.iv_tx_data.iv_transid.id[0]));
               abort ();
            }
            break;
        }
        case TM_UP:
        {
            break; 
        }
        case TM_STATE_RESYNC:
        {
             TMTrace(3, ("tm_sync_cb - TM_STATE_RESYNC received \n"));
             // nothing to validate since these are booleans and the node
             // being recovered could be a -1
             break;
        }
        case TM_RECOVERY_START:
        case TM_RECOVERY_END:
        {
            if ((lp_data->u.iv_to_data.iv_my_node < 0) ||
                 (lp_data->u.iv_to_data.iv_my_node > MAX_NODES) ||
                 (lp_data->u.iv_to_data.iv_down_node < 0) ||
                 (lp_data->u.iv_to_data.iv_down_node > MAX_NODES))
            {
                tm_log_event(DTM_TM_NODE_OUTSIDE_RANGE, SQ_LOG_CRIT, "DTM_TM_NODE_OUTSIDE_RANGE");
                TMTrace(1, ("tm_sync_cb - Received RECOVERY sync with node out of range.\n"));
                abort ();
            }


#ifdef DEBUG_MODE
           bool lv_assert = false;
           ms_getenv_bool("TM_TEST_AFTER_REC_START_SYNC_ASSERT", &lv_assert);
           if (lv_assert == true)
           {
              // sprintf(la_buf, "TM Test: Rec start sync assert\n");
            //   tm_log_write(DTM_TM_TEST_REC_START_SYNC, SQ_LOG_CRIT, la_buf);
              // TMTrace(1, ("tm_sync_cb - %s", la_buf));
               abort ();
           }
           
#endif 
             break;
        }
        case TM_LISTBUILT_SYNC:
        {
            if ((lp_data->u.iv_list_built.iv_down_node < 0) ||
                 (lp_data->u.iv_list_built.iv_down_node > MAX_NODES))
            {
                tm_log_event(DTM_TM_NODE_OUTSIDE_RANGE, SQ_LOG_CRIT, "DTM_TM_NODE_OUTSIDE_RANGE");
                TMTrace(1, ("tm_sync_cb - Received TM_LISTBUILT_SYNC sync with node out of range\n"));
                abort ();
            }
            TMTrace(1, ("tm_sync_cb - received TM_LISTBUILT_SYNC, verification successful\n"));
            break; 
        }
        case TM_PROCESS_RESTART:
            break;
        case TM_SYS_RECOV_START_SYNC:
        case TM_SYS_RECOV_END_SYNC:
        {
            if ((lp_data->u.iv_sys_recov_data.iv_sys_recov_state > TM_SYS_RECOV_STATE_END) ||
                 (lp_data->u.iv_sys_recov_data.iv_sys_recov_lead_tm_node < 0) ||
                 (lp_data->u.iv_sys_recov_data.iv_sys_recov_lead_tm_node > MAX_NODES))
            {
                tm_log_event(DTM_TM_NODE_OUTSIDE_RANGE, SQ_LOG_CRIT, "DTM_TM_NODE_OUTSIDE_RANGE");
                TMTrace(1, ("tm_sync_cb - Received RECOVERY sync with node out of range\n"));
                abort ();
            }

            break;
        }
        default:
        {
            tm_log_event(DTM_TM_UNKNOWN_SYNC_TYPE, SQ_LOG_CRIT, "DTM_TM_UNKNOWN_SYNC_TYPE");
            TMTrace(1, ("tm_sync_cb - Unknown sync header type received\n"));
            abort ();
            break;
        }
    }; 

    memcpy (lp_sync_data, lp_data, sizeof (Tm_Sync_Data));
    gv_sync_map.put(pv_handle, lp_sync_data);

    TMTrace(2, ("tm_sync_cb EXIT : type %d\n", lp_hdr->iv_type));

    return 0;
}


void tm_recipient_sync_commit (Tm_Sync_Data *pp_sync_data)
{
    TMTrace(2, ("tm_recipient_sync_commit : ENTRY, type %d\n",
                                    pp_sync_data->iv_hdr.iv_type));
    switch (pp_sync_data->iv_hdr.iv_type)
    {
        case TM_BEGIN_SYNC:
        {
            gv_tm_info.add_sync_data(pp_sync_data->iv_hdr.iv_nid, 
                                     &pp_sync_data->u.iv_tx_data);
            gv_system_tx_count++;
            break;  
        }
        case TM_END_SYNC:
        {

            Tm_Tx_Sync_Data *lp_data = gv_tm_info.get_sync_data(
                                       &pp_sync_data->u.iv_tx_data);
            // Add sync data to the sync data list if
            // it isn't already in the list for the sending node.
            if (lp_data == NULL)
               gv_tm_info.add_sync_data(pp_sync_data->iv_hdr.iv_nid, 
                                      &pp_sync_data->u.iv_tx_data);
            else
               lp_data->iv_state = pp_sync_data->u.iv_tx_data.iv_state;

            break;
        }
        case TM_FORGET_SYNC:
        {
            gv_tm_info.remove_sync_data(&pp_sync_data->u.iv_tx_data);
            break;
        }
        case TM_STATE_RESYNC:
        {

            TMTrace(3, ("tm_recipient_sync_commit - TM_STATE_RESYNC sync received.\n"));
            gv_tm_info.node_being_recovered(pp_sync_data->u.iv_state_resync.iv_index,
                                            pp_sync_data->u.iv_state_resync.iv_node_being_recovered);
            gv_tm_info.down_without_sync( pp_sync_data->u.iv_state_resync.iv_index,
                                            pp_sync_data->u.iv_state_resync.iv_down_without_sync);
            gv_tm_info.recovery_list_built( pp_sync_data->u.iv_state_resync.iv_index,
                                            pp_sync_data->u.iv_state_resync.iv_list_built);
            break;
        }
        case TM_RECOVERY_START:
        {
            tm_log_event(DTM_TM_START_NODE_RECOVERY, SQ_LOG_INFO, "DTM_TM_START_NODE_RECOVERY");
            TMTrace(1, ("tm_recipient_sync_commit - RECOVERY START sync received.\n"));
            // The lead TM can not receive this sync. Issue an event and shutdown the cluster
            if (gv_tm_info.lead_tm())
            {
               tm_log_event(DTM_LEAD_TM_TM_SYNC_UNEXPECTED, SQ_LOG_CRIT, "DTM_LEAD_TM_TM_SYNC_UNEXPECTED", FEDUP);
               TMTrace(1, ("tm_recipient_sync_recipient : Error TM_RECOVERY_START sync received by Lead TM.\n"));
               gv_tm_info.error_shutdown_abrupt(FEDUP);
            }
            if (pp_sync_data->u.iv_to_data.iv_down_node == -1)
            {
                tm_log_event(DTM_TM_START_NODE_RECOVERY, SQ_LOG_CRIT, "DTM_TM_START_NODE_RECOVERY");
                TMTrace(1, ("tm_recipient_sync_commit - Invalid node id received for a RECOVERY START sync\n"));
                abort ();
            }
               
            gv_tm_info.node_being_recovered (
                       pp_sync_data->u.iv_to_data.iv_down_node,
                       pp_sync_data->u.iv_to_data.iv_my_node);
            gv_tm_info.down_without_sync(pp_sync_data->u.iv_to_data.iv_down_node, false);
            TMTrace(3, ("tm_recipient_sync_commit - setting down_without_sync to FALSE for node %d\n",
                          pp_sync_data->u.iv_to_data.iv_down_node));
            gv_tm_info.schedule_init_and_recover_rms();
            break;
        }
        case TM_RECOVERY_END:
        {
            tm_log_event(DTM_TM_END_NODE_RECOVERY, SQ_LOG_INFO, "DTM_TM_END_NODE_RECOVERY");
            TMTrace(1, ("tm_recipient_sync_commit - RECOVERY END sync received.\n"));
            if (pp_sync_data->u.iv_to_data.iv_down_node == -1)
            {
                tm_log_event(DTM_TM_END_NODE_RECOVERY, SQ_LOG_CRIT, "DTM_TM_END_NODE_RECOVERY");
                TMTrace(1, ("tm_recipient_sync_commit - Invalid node id received for a RECOVERY END sync.\n"));
                abort ();
            }
                
            //reset
            TMTrace(3, ("tm_recipient_sync_commit setting recovery_list_built to FALSE for Node " 
                         " %d\n",pp_sync_data->u.iv_to_data.iv_down_node));
            // reset list built flag for recovery
            gv_tm_info.recovery_list_built (pp_sync_data->u.iv_to_data.iv_down_node, false);
            gv_tm_info.node_being_recovered (pp_sync_data->u.iv_to_data.iv_down_node, -1);
            gv_tm_info.set_sys_recov_status(TM_SYS_RECOV_STATE_END,
                        pp_sync_data->u.iv_to_data.iv_my_node); //my node sb the lead tm
            gv_tm_info.tm_up();
            break;
        }
        case TM_LISTBUILT_SYNC:
        {
           tm_log_event(DTM_TM_LISTBUILT_SYNC, SQ_LOG_INFO, "DTM_TM_LISTBUILT_SYNC");
            TMTrace(3, ("tm_recipient_sync_commit (TM_LISTBUILT_SYNC) setting recovery_list_built " 
                        " to TRUE for Node %d\n",pp_sync_data->u.iv_list_built.iv_down_node));
            gv_tm_info.recovery_list_built (pp_sync_data->u.iv_list_built.iv_down_node, true);
            break;
        }
        case TM_PROCESS_RESTART:
           tm_log_event(DTM_TM_PROCESS_RESTART_SYNC, SQ_LOG_INFO, "DTM_TM_PROCESS_RESTART_SYNC");
           TMTrace(1, ("tm_recipient_sync_commit - process restart sync received.\n"));

           gv_tm_info.restarting_tm(pp_sync_data->u.iv_proc_restart_data.iv_proc_restart_node);
           gv_tm_info.schedule_init_and_recover_rms();
           break;
        case TM_SYS_RECOV_START_SYNC:
        case TM_SYS_RECOV_END_SYNC:
        {
            gv_tm_info.set_sys_recov_status(pp_sync_data->u.iv_sys_recov_data.iv_sys_recov_state,
                        pp_sync_data->u.iv_sys_recov_data.iv_sys_recov_lead_tm_node);
            break;
        }       
        case TM_UP:
        {
            // this is received upon startup after recovery, so its a fresh system
            gv_tm_info.can_takeover(true);
            gv_tm_info.tm_up();
            break;
        }
        default:
        {
            tm_log_event(DTM_TM_UNKNOWN_SYNC_TYPE, SQ_LOG_CRIT, "DTM_TM_UNKNOWN_SYNC_TYPE");
            TMTrace(1, ("tm_recipient_sync_commit : invalid data\n"));

            abort ();
            break;
        }
    };
    TMTrace(2, ("tm_recipient_sync_commit EXIT \n"));
}

// --------------------------------------------------------------------
// tm_get_leader_info
// Purpose : Get the new tm leader. 
// --------------------------------------------------------------------
void tm_get_leader_info()
{
    // Nothing to do here if we are already the Lead TM.
    if (gv_tm_info.lead_tm() == true)
         return;

    int32 lv_leader_nid, lv_leader_pid;
    char la_leader_name[BUFSIZ]; 
    int32 lv_old_leader_nid = gv_tm_info.lead_tm_nid();

    int lv_leader_error = msg_mon_tm_leader_set(&lv_leader_nid, 
                                            &lv_leader_pid, la_leader_name);
    // ignore error as it simply indicates that we are not the leader.
    if (lv_leader_error)
    {
        TMTrace(3, ("tm_get_leader_info : Error %d returned by "
                "msg_mon_tm_leader_set - $TM%d is not the Lead. Error ignored.\n", 
                lv_leader_error, gv_tm_info.nid()));
    }
    gv_tm_info.lead_tm_nid(lv_leader_nid);

    if (lv_leader_nid != lv_old_leader_nid)
    {
         tm_log_event (DTM_TM_LEADTM_SET, SQ_LOG_INFO , "DTM_TM_LEADTM_SET", 
                    -1, /*error_code*/ 
                    -1, /*rmid*/
                    gv_tm_info.nid(), /*dtmid*/ 
                    -1, /*seq_num*/
                    -1, /*msgid*/
                    -1, /*xa_error*/
                    -1, /*pool_size*/
                    -1, /*pool_elems*/
                    -1, /*msg_retries*/
                    -1, /*pool_high*/
                    -1, /*pool_low*/
                    -1, /*pool_max*/
                    -1, /*tx_state*/
                    lv_old_leader_nid, /*data */
                    -1, /*data1*/
                    -1,/*data2 */
                    NULL, /*string2*/
                    lv_leader_nid /*node*/);
         TMTrace(3, ("tm_get_leader_info : Node %d is new Lead DTM.\n", lv_leader_nid));

         if (lv_leader_nid == gv_tm_info.nid())
         {
            // modify the wait interval now for this lead dtm
            gv_wait_interval = LEAD_DTM_WAKEUP_INTERVAL/10;
            gv_tm_info.lead_tm(true);
            gv_tm_info.lead_tm_takeover(true);
            gv_tm_info.open_other_tms();
            // Add a Checkpoint event to drive cp processing
//            gv_tm_info.tmTimer()->cancelControlpointEvent();
//            gv_tm_info.tmTimer()->addControlpointEvent(gv_tm_info.cp_interval());
            // Add a stats event
            gv_tm_info.tmTimer()->cancelStatsEvent();
            gv_tm_info.tmTimer()->addStatsEvent(gv_tm_info.stats_interval());
         }

    }
    else
    {
         TMTrace(3, ("tm_get_leader_info : Lead DTM did not change. Node %d is still Lead DTM.\n", 
                 lv_leader_nid));
    }

} //tm_get_leader_info

//---------------------------------------------------------------------
// tm_originating_sync_commit
// Purpose - helper method to process the phase2 sync from the 
//           originating TM
// --------------------------------------------------------------------
void tm_originating_sync_commit (int32 pv_tag)
{
    CTmTxBase *lp_tx             = NULL;
    Tm_Sync_Type_Transid *lp_data = gv_tm_info.get_sync_otag(pv_tag);
    CTmTxMessage *lp_msg;

    // assert (lp_data != NULL);
    if (lp_data == NULL)
    {
       TMTrace(1, ("tm_originating_sync_commit : ERROR tag %d not found in sync tags, sync ignored.\n",
                     pv_tag));
        return;
    }

    TMTrace(2, ("tm_originating_sync_commit ENTRY, ID %d, tag %d, type %d.\n",
                     lp_data->u.iv_seqnum, pv_tag, lp_data->iv_sync_type));

    switch(lp_data->iv_sync_type)
    {
        case TM_END_SYNC:
        case TM_BEGIN_SYNC:
        {
            lp_tx = (CTmTxBase *) gv_tm_info.get_tx(lp_data->u.iv_node_to_takeover, lp_data->u.iv_seqnum);

            if (lp_tx == NULL)
            {
                tm_log_event(DTM_TM_INVALID_TRANSACTION, SQ_LOG_CRIT, "DTM_TM_INVALID_TRANSACTION");
                TMTrace(1, ("tm_originating_sync_commit : END/BEGIN SYNC - Unable to find "
                            " transaction during a phase 2 sync\n"));
                abort ();
            }
            
            TMTrace(3, ("tm_originating_sync_commit : END/BEGIN SYNC "
                              "for ID %d, Tx.SeqNum %d, Tx.Tag %d, tag %d, type %d.\n",
                              lp_data->u.iv_seqnum, lp_tx->seqnum(), lp_tx->tag(), pv_tag, lp_data->iv_sync_type));
            lp_tx->schedule_eventQ(); 
            break;
         }
         case TM_FORGET_SYNC:
         {
            // we are done, queue endforget event against txn thread.
            lp_tx = (CTmTxBase *) gv_tm_info.get_tx(lp_data->u.iv_node_to_takeover, lp_data->u.iv_seqnum);

            // If the transaction object doesn't exist we assume that it's already been
            // cleaned up.  This can happen, for example, when a begin or end sync completion
            // arrives after we've already issued the forget sync.  TM_TX_Info::schedule_eventQ
            // will drive forget processing because the transaction is in forgotten state.
            // In this case the best thing we can do is simply through away the forget sync
            // completion here.
            if (lp_tx == NULL)
            {
                tm_log_event(DTM_TM_INVALID_TRANSACTION, SQ_LOG_CRIT, "DTM_TM_INVALID_TRANSACTION");
                TMTrace(1, ("tm_originating_sync_commit : FORGET_SYNC - WARNING "
                            "Unable to find transaction for a phase 2 "
                            "forget sync completion.  Completion assumed out of order and ignored!\n"));
                abort ();
            }            
            else
            {
               TMTrace(3, ("tm_originating_sync_commit : FORGET_SYNC "
                              "for ID %d, Tx.SeqNum %d, Tx.Tag %d, tag %d, type %d.\n",
                              lp_data->u.iv_seqnum, lp_tx->seqnum(), lp_tx->tag(), pv_tag, lp_data->iv_sync_type));

               lp_msg = lp_tx->currRequest();
               if (lp_msg)
               {
                  lp_msg->requestType(TM_MSG_TXINTERNAL_ENDFORGET);
                  lp_tx->eventQ_push(lp_msg);
               }
               else
               {
                  tm_log_event(DTM_TM_INVALID_TRANSACTION, SQ_LOG_CRIT, "DTM_TM_INVALID_TRANSACTION");
                  TMTrace(1, ("tm_originating_sync_commit : FORGET_SYNC - Forget Sync phase 2 for transaction "
                         "%d but request "
                        "has already completed!  Forget ignored.\n", lp_tx->seqnum()));
               }            
            }
            break;
          }
          case TM_STATE_RESYNC:
          {
              TMTrace(1, ("tm_originating_sync_commit, TM_STATE_RESYNC received, no-op\n"));
              break;
          }
          case TM_RECOVERY_START:
          {       
#ifdef DEBUG_MODE
             bool lv_verify = false;
             ms_getenv_bool("TM_VERIFY", &lv_verify);
             if (lv_verify)
             {
                if (gv_tm_info.iv_trace_level)
                {
                   if (gv_tm_info.tm_test_verify(lp_data->u.iv_node_to_takeover))
                       trace_printf("tm_verify after takeover successful\n");
                   else
                       trace_printf("tm_verify after takeover ERROR\n");
                }
             }
#endif
             break;
          }
          case TM_RECOVERY_END:
          {   
               break;
          }
          case TM_LISTBUILT_SYNC:
          {
               // We don't need to do anything in the originating TM since we've already 
               // recorded the appropriate flags.
               TMTrace(3, ("tm_originating_sync_commit : received TM_LISTBUILT_SYNC, no-op.\n"));
               break;
          }
          case TM_UP:
          {
              // set registry entry to indicate that transaction service is ready.
              gv_tm_info.set_txnsvc_ready(TXNSVC_UP);
              break;
              
           }
          case TM_PROCESS_RESTART:
              break;
           case TM_SYS_RECOV_START_SYNC:
           {
               // software fault
               if (!gv_tm_info.ClusterRecov())
               {
                    tm_log_event(DTM_RECOVERY_FAILED2, SQ_LOG_CRIT, "DTM_RECOVERY_FAILED2",
                                  -1,-1,gv_tm_info.nid());
                    abort(); // this is a software fault that doesn't warrant taking
                             // down the cluster
               }

               // System recovery now runs in the timer thread to keep the main thread from blocking.
               gv_tm_info.schedule_recover_system();      
               break;
           }
           case TM_SYS_RECOV_END_SYNC:
           {
               // Lead TM: Send out TM_UP sync and then set TM_UP to allow new transactions to be processed.
               send_state_up_sync(gv_tm_info.nid());
               gv_tm_info.tm_up();
               gv_tm_info.can_takeover(true);
               break;
           }
           default:
           {
               // TODO
               break;
            } 
       };
    
    TMTrace(2, ("tm_originating_sync_commit EXIT, TxnId %d, sync type %d\n",
                     lp_data->u.iv_seqnum, lp_data->iv_sync_type));
    gv_tm_info.remove_sync_otag(pv_tag);
}

void tm_originating_sync_abort(int32 pv_tag)
{
     CTmTxBase *lp_tx             = NULL;
     Tm_Sync_Type_Transid *lp_data = gv_tm_info.get_sync_otag(pv_tag);


     // we need to allow this to not be here as the monitor can choose the abort...
     if (lp_data == NULL)
     {
          TMTrace(1, ("tm_originating_sync_abort : NULL data for tag %d\n", pv_tag));
     }
     else
     {
         TMTrace(2, ("tm_originating_sync_abort ENTRY, tag=%d, type=%d\n", pv_tag, lp_data->iv_sync_type));

         if (lp_data->iv_num_tries >=3 )
         {
              tm_log_event(DTM_TM_EXCEEDED_SYNC_ABORT_TRIES, SQ_LOG_CRIT, "DTM_TM_EXCEEDED_SYNC_ABORT_TRIES");
              TMTrace(1, ("tm_originating_sync_abort : max number of retries, exiting.\n"));

              abort (); // retry 3 times
         }
         
         lp_data->iv_num_tries++;

         switch(lp_data->iv_sync_type)
         {
             case TM_BEGIN_SYNC:
             case TM_END_SYNC:
             case TM_FORGET_SYNC:
             {
                 lp_tx = (CTmTxBase *) gv_tm_info.get_tx(lp_data->u.iv_node_to_takeover, lp_data->u.iv_seqnum);

                 if (lp_tx == NULL)
                 {
                     tm_log_event(DTM_TM_INVALID_TRANSACTION, SQ_LOG_CRIT, "DTM_TM_INVALID_TRANSACTION");
                     TMTrace(1, ("tm_originating_sync_abort - Unable to find transaction "
                                 "during a phase 2 sync.\n"));
                     abort ();
                 }
                 
                 lp_tx->schedule_redrive_sync();
                 break;
             }
             case TM_UP:
            {
                  send_state_up_sync(gv_tm_info.nid()); 
                  break;
               
            }
            case TM_STATE_RESYNC:
            {
                TMTrace(3, ("tm_originating_sync_abort - TM_STATE_RESYNC sync received.\n"));
                send_state_resync (gv_tm_info.nid(),
                                       gv_tm_info.down_without_sync(lp_data->u.iv_node_to_takeover),
                                       gv_tm_info.node_being_recovered(lp_data->u.iv_node_to_takeover),
                                       gv_tm_info.recovery_list_built(lp_data->u.iv_node_to_takeover), 
                                       lp_data->u.iv_node_to_takeover);
                break;
            }
             case TM_RECOVERY_START:
             {
                 send_takeover_tm_sync (TM_RECOVERY_START, gv_tm_info.nid(), 
                                        lp_data->u.iv_node_to_takeover);
                 break;
             } 
             case TM_RECOVERY_END:
             {
                 send_takeover_tm_sync (TM_RECOVERY_END, gv_tm_info.nid(), 
                                       lp_data->u.iv_node_to_takeover);                         
                 break;
             } 
             case TM_LISTBUILT_SYNC:
             {
                 send_recov_listbuilt_sync (gv_tm_info.nid(), lp_data->u.iv_node_to_takeover);
                 break;

             }
             case TM_PROCESS_RESTART:
                 break;
              case TM_SYS_RECOV_START_SYNC:
              {
                  send_sys_recov_start_sync(gv_tm_info.nid()); 
                  break;
              }
              case TM_SYS_RECOV_END_SYNC:
              {
                  send_sys_recov_end_sync(gv_tm_info.nid()); 
                  break;
              }
              default:
              {
                 tm_log_event(DTM_TM_UKN_SYNC_TYPE, SQ_LOG_WARNING, "DTM_TM_UKN_SYNC_TYPE");
                 break;
              } 
         };
         // do not remove from table as we will retry
     }
     TMTrace(2, ("tm_originating_sync_abort EXIT\n"));
}


// ---------------------------------------------------------------------------
// tm_process_node_down_msg
// Purpose : Process a down msg from the monitor (virtual nodes). For real
// clusters, process a DTM process death in the case of a logical node failure
// ---------------------------------------------------------------------------
void tm_process_node_down_msg(int32 pv_nid)
{
    gv_tm_info.close_tm(pv_nid); 
    TMTrace(2, ("tm_process_node_down_msg ENTRY, nid %d\n", pv_nid));
    tm_log_event(DTM_NODEDOWN, SQ_LOG_INFO, "DTM_NODEDOWN", 
        -1,-1,gv_tm_info.nid(),-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,
        NULL,pv_nid);

    if (!gv_tm_info.lead_tm())
    {
        tm_get_leader_info();
        if (gv_tm_info.lead_tm() == false)
        {
            TMTrace(2, ("tm_process_node_down_msg EXIT - %d is not the lead TM.\n", pv_nid));
            return;
        }
    }

    if ((gv_tm_info.sys_recov_state() != TM_SYS_RECOV_STATE_END) && 
        (gv_tm_info.sys_recov_lead_tm_nid() == pv_nid))
        // If this is system startup time and system recovery has not yet ended
        // and the down node is the previous lead TM node, this new Lead TM needs
        // to perform system recovery again.  There are no new outstanding 
        // transactions to take over from the down node at this stage since 
        // transaction has now yet been enabled.  
    {
        gv_tm_info.ClusterRecov(new TM_Recov(gv_tm_info.rm_wait_time()));
        gv_tm_info.ClusterRecov()->initiate_start_sync();
    }
    else 
    {       
        if (gv_tm_info.can_takeover())
        {
            // take over phase 1 is now called by TM_Info:restart_tm to make sure
            // it happens after the tm process starts.
            //tm_process_take_over_phase1 (pv_nid);
            // lets start with a clean slate and write a control
            // point after a takeover just in case the lead went
            // down, and for an otherwise fresh start
            if ((gv_tm_info.state() == TM_STATE_SHUTTING_DOWN) ||
                (gv_tm_info.state() == TM_STATE_SHUTDOWN_COMPLETED))
            {
               if (gv_tm_info.num_active_txs() <= 0)
               {
                  // redrive the shutdown operation
                  TMShutdown *lp_Shutdown = new TMShutdown(&gv_tm_info, gv_RMs.TSE()->return_rms());
                  gv_tm_info.shutdown_coordination_started(true);
                  lp_Shutdown->coordinate_shutdown();
                  delete lp_Shutdown;
                  // This must be the lead TM.  After the shutdown, set the registry
                  // entry to indicate that transaction service has stoppped
                  gv_tm_info.set_txnsvc_ready(TXNSVC_DOWN);
               }
            }
        }
    }
    // Since the node is down, the TM is closed
    gv_tm_info.close_tm(pv_nid);
    TMTrace(2, ("tm_process_node_down_msg EXIT nid %d\n", pv_nid));
} //tm_process_node_down_msg


// -----------------------------------------------------------------
// tm_process_node_quiesce_msg
// Purpose : process a quiesce node notice from the Monitor.
// This can be received by any TM.  The TM suspends transaction
// processing but will still process Management requests and TSE
// replies to outstanding requests.  The Monitor will kill this
// TM process once TSEs have completed control pointing.
// pv_stop is only set to true for TM testing.
// ----------------------------------------------------------------
void tm_process_node_quiesce_msg(CTmTxMessage *pp_msg=NULL)
{
   short lv_error = FEOK;
   static int32 lv_lastTMState = gv_tm_info.state();
   bool lv_stop = (pp_msg)?pp_msg->request()->u.iv_quiesce.iv_stop:false;

   TMTrace(2, ("tm_process_node_quiesce_msg ENTRY, stop=%d, current TM State %d.\n",
           lv_stop, lv_lastTMState));
    tm_log_event(DTM_NODEQUIESCE, SQ_LOG_INFO, "DTM_NODEQUIESCE", 
        -1,-1,gv_tm_info.nid(),-1,-1,-1,-1,-1,-1,-1,-1,-1,gv_tm_info.state(),lv_stop);

   if (lv_stop)
   {
      if (gv_tm_info.state() != TM_STATE_QUIESCE)
      {
          TMTrace(1, ("tm_process_node_quiesce_msg - Must quiesce first!!\n"));
      }
      else
          gv_tm_info.state(lv_lastTMState);
   }
   else
      gv_tm_info.state(TM_STATE_QUIESCE);

   tm_log_event(DTM_TM_QUIESCED, SQ_LOG_WARNING, "DTM_TM_QUIESCED",
                -1, -1, gv_tm_info.nid());
   TMTrace(1, ("TM %d quiescing.\n", gv_tm_info.nid()));
   if (pp_msg != NULL)
   {
      pp_msg->reply(lv_error);
      delete pp_msg;
   }
   TMTrace(2, ("tm_process_node_quiesce_msg EXIT.\n"));
} //tm_process_req_quiesce


void tm_abort_all_transactions(bool pv_shutdown)
{
    TMTrace(2, ("tm_abort_all_transactions ENTRY with shutdown=%d.\n", pv_shutdown));

    gv_tm_info.abort_all_active_txns();

    if (!pv_shutdown)
       gv_tm_info.state(TM_STATE_UP);

    TMTrace(2, ("tm_abort_all_transactions EXIT\n"));
}
// ----------------------------------------------------------
// tm_process_registry_change
// Purpose - determine if a DTM key was changed and if we need 
//           to take action 
// -----------------------------------------------------------
void tm_process_registry_change(MS_Mon_Change_def *pp_change )
{
    int32 lv_value; 
    char lv_regKeyText[1024];
    char *lp_regKeyText = (char *) &lv_regKeyText;

    sprintf(lp_regKeyText, "%s:%s=%s", pp_change->group, 
            pp_change->key, pp_change->value);
    TMTrace(1, ("tm_process_registry_change Registry Change notice key %s.\n", lp_regKeyText));
    //tm_log_event(DTM_REGCHANGE_NOTICE, SQ_LOG_INFO, "DTM_REGCHANGE_NOTICE",
    //            -1,-1,gv_tm_info.nid(),-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,lp_regKeyText);

    if (strcmp(pp_change->key, DTM_STALL_PHASE_2) == 0)
    {
       lv_value = atoi (pp_change->value);
       if (lv_value >= 0)
       {
          gv_tm_info.stall_phase_2(lv_value);
          HbaseTM_initiate_stall(lv_value);
       }
    }
    else if (strcmp(pp_change->key, DTM_RM_WAIT_TIME) == 0)
    {
       lv_value = atoi (pp_change->value);
       if (lv_value > 0)
       {
          lv_value *= 100; //  100 (secs to 10 msecs)
          gv_tm_info.rm_wait_time(lv_value);
       }
    }
    else if (strcmp(pp_change->key, DTM_TM_TRACE) == 0)
    {
        lv_value = atoi (pp_change->value);
        if (lv_value >= 0)
            gv_tm_info.set_trace(lv_value/*detail*/);
    }
    else if (strcmp(pp_change->key, DTM_TRANS_HUNG_RETRY_INTERVAL) == 0)
    {
        lv_value = atoi (pp_change->value);
        if (lv_value > 0)
          gv_tm_info.trans_hung_retry_interval(lv_value);
    }
    else if (strcmp(pp_change->key, DTM_XATM_TRACE) == 0)
    {
        if (strcmp(pp_change->value,"") != 0)
            gv_tm_info.set_xa_trace(pp_change->value);
    }
    else if (strcmp(pp_change->key, DTM_TIMERTHREAD_WAIT) == 0)
    {
        lv_value = atoi (pp_change->value);
        if (lv_value > 0 || lv_value == -1)
          gv_tm_info.timerDefaultWaitTime(lv_value);
        gv_tm_info.tmTimer()->defaultWaitTime(gv_tm_info.timerDefaultWaitTime());
    }
    // Note that with pool configuration parameters you must set/alter them in a 
    // specific order if they overlap because they are parsed separately.
    // Increasing values: Set max, then ss_high, then ss_low.
    // Decreasing values: Set ss_low, then ss_high, then max.
    else if (strcmp(pp_change->key, DTM_TM_STATS) == 0)
    {
        lv_value = atoi (pp_change->value);
        bool lv_tm_stats = ((lv_value == 0)?false:true);
        gv_tm_info.stats()->initialize(lv_tm_stats, gv_tm_info.stats()->collectInterval());
        gv_tm_info.threadPool()->setConfig(lv_tm_stats);
        // Add other pools here
    }
    // Configure thread pool
    else if (strcmp(pp_change->key, DTM_MAX_NUM_THREADS) == 0)
    {
        lv_value = atoi (pp_change->value);
        if (lv_value >= 1)
           gv_tm_info.threadPool()->setConfig(gv_tm_info.tm_stats(), lv_value);
    }
    else if (strcmp(pp_change->key, DTM_STEADYSTATE_LOW_THREADS) == 0)
    {
        lv_value = atoi (pp_change->value);
        if (lv_value >= 0)
           gv_tm_info.threadPool()->setConfig(gv_tm_info.tm_stats(), -1, lv_value);
    }
    else if (strcmp(pp_change->key, DTM_STEADYSTATE_HIGH_THREADS) == 0)
    {
        lv_value = atoi (pp_change->value);
        if (lv_value >= 0)
           gv_tm_info.threadPool()->setConfig(gv_tm_info.tm_stats(), -1, -1, lv_value);
    }
    // Configure transaction pool
    else if (strcmp(pp_change->key, DTM_MAX_NUM_TRANS) == 0)
    {
        lv_value = atoi (pp_change->value);
        if (lv_value >= 1)
           gv_tm_info.transactionPool()->setConfig(gv_tm_info.tm_stats(), lv_value);
    }
    else if (strcmp(pp_change->key, DTM_STEADYSTATE_LOW_TRANS) == 0)
    {
        lv_value = atoi (pp_change->value);
        if (lv_value >= 0)
           gv_tm_info.transactionPool()->setConfig(gv_tm_info.tm_stats(), -1, lv_value);
    }
    else if (strcmp(pp_change->key, DTM_STEADYSTATE_HIGH_TRANS) == 0)
    {
        lv_value = atoi (pp_change->value);
        if (lv_value >= 0)
           gv_tm_info.transactionPool()->setConfig(gv_tm_info.tm_stats(), -1, -1, lv_value);
    }
    else if (strcmp(pp_change->key, DTM_CP_INTERVAL) == 0)
    {
        lv_value = atoi (pp_change->value);
        if (lv_value > 0)
         lv_value *= 60000; // 60 (mins to secs) * 1000 (secs to msecs)
        if (lv_value >= 0 && lv_value != gv_tm_info.cp_interval())
        {
          // Cancel the TmTimer control point event and re-add with the
          // new interval.
          gv_tm_info.cp_interval(lv_value);
          gv_tm_info.tmTimer()->cancelControlpointEvent();
          gv_tm_info.tmTimer()->addControlpointEvent(lv_value);
        }
    }
    else if (strcmp(pp_change->key, DTM_STATS_INTERVAL) == 0)
    {
        lv_value = atoi (pp_change->value);
        if (lv_value > 0)
         lv_value *= 60000; // 60 (mins to secs) * 1000 (secs to msecs)
        if (lv_value >= 0 && lv_value != gv_tm_info.stats_interval())
        {
            // Cancel the TmTimer stats event and re-add with the
            // new interval.
            gv_tm_info.stats_interval(lv_value);
            gv_tm_info.tmTimer()->cancelStatsEvent();
            gv_tm_info.tmTimer()->addStatsEvent(lv_value);
        }
    }
    else if (strcmp(pp_change->key, DTM_TM_RMRETRY_INTERVAL) == 0)
    {
        lv_value = atoi (pp_change->value);
        if (lv_value > 0)
         lv_value *= 60000; // 60 (mins to secs) * 1000 (secs to msecs)
        if (lv_value >= 0 && lv_value != gv_tm_info.RMRetry_interval())
        {
            // Cancel the TmTimer stats event and re-add with the
            // new interval.
            gv_tm_info.RMRetry_interval(lv_value);
            gv_tm_info.tmTimer()->cancelRMRetryEvent();
            gv_tm_info.tmTimer()->addRMRetryEvent(lv_value);
        }
    }
    else if (strcmp(pp_change->key, DTM_TX_ABORT_TIMEOUT) == 0)
    {
       lv_value = atoi (pp_change->value);
       if (lv_value != -1 && lv_value <= 0)
          gv_tm_info.timeout(TX_ABORT_TIMEOUT); //Default
       else
          gv_tm_info.timeout(lv_value);
    }
    else if (strcmp(pp_change->key, DTM_TEST_PAUSE_STATE) == 0)
    {
       lv_value = atoi (pp_change->value);
       if (lv_value < TM_TX_STATE_NOTX || lv_value > TM_TX_STATE_LAST)
       {
          if (lv_value == -2)
          {
             TMTrace(1,("DTM_TEST_PAUSE_STATE set to %d, type %d = random!\n",  
                     lv_value, gv_pause_state_type));
             srand(time(NULL));
             gv_pause_state_type = TX_PAUSE_STATE_TYPE_RANDOM;
             gv_pause_state = rand() % TM_TX_STATE_LAST; //starting point
          }
          else
          {
             TMTrace(1,("DTM_TEST_PAUSE_STATE set to default (-1) because %d not a value state, type %d.\n", 
                     lv_value, TM_TX_STATE_NOTX));
             gv_pause_state = -1; //Default
          }
       }
       else
       {
          TMTrace(1,("DTM_TEST_PAUSE_STATE set to %d, type %d.\n",  lv_value, gv_pause_state_type));
          gv_pause_state = lv_value;
       }
    }
    else if (strcmp(pp_change->key, DTM_RM_PARTIC) == 0)
    {
       lv_value = atoi (pp_change->value);
       gv_tm_info.RMPartic(lv_value);
       TMTrace (1, ("DTM_RM_PARTIC set to %d.\n", gv_tm_info.RMPartic()));
    }
    else if (strcmp(pp_change->key, DTM_TM_TS_MODE) == 0)
    {
       lv_value = atoi (pp_change->value);
       gv_tm_info.TSMode((TS_MODE) lv_value);
       TMTrace (1, ("DTM_TM_TS_MODE set to %d.\n", gv_tm_info.TSMode()));
    }
    else if (strcmp(pp_change->key, DTM_TM_SHUTDOWNABRUPTNOW) == 0)
    {
       lv_value = atoi (pp_change->value);
       if (lv_value == 1)
       {
          TMTrace (1, ("DTM_TM_SHUTDOWNABRUPTNOW set, calling shutdown, abrupt. Use for testing only!!\n"));
          tm_log_event(DTM_ERROR_SHUTDOWN_DEBUG, SQ_LOG_INFO, "DTM_ERROR_SHUTDOWN_DEBUG");
          msg_mon_shutdown(MS_Mon_ShutdownLevel_Abrupt);
       }
    }
    else if (strcmp(pp_change->key, DTM_BROADCAST_ROLLBACKS) == 0)
    {
        bool lv_changed = false;
        TM_BROADCAST_ROLLBACKS lv_broadcast_rollbacks = (TM_BROADCAST_ROLLBACKS) atoi (pp_change->value);
        switch (lv_broadcast_rollbacks)
        {
        case TM_BROADCAST_ROLLBACKS_NO:
        case TM_BROADCAST_ROLLBACKS_YES:
        case TM_BROADCAST_ROLLBACKS_DEBUG:
           gv_tm_info.broadcast_rollbacks(lv_broadcast_rollbacks);
           lv_changed = true;
           break;
        }
        if (lv_changed)
        {
           tm_log_event (DTM_BROADCAST_ROLLBACKS_INFO, SQ_LOG_INFO,"DTM_BROADCAST_ROLLBACKS_INFO",
                         -1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,gv_tm_info.broadcast_rollbacks()); 
           TMTrace(1, ("DTM_BROADCAST_ROLLBACKS changed to %d.\n", gv_tm_info.broadcast_rollbacks()));
        }
    }

}

// tm_process_monitor_msg
// Purpose - when a monitor message is received, this is called
// ------------------------------------------------------------ 
void tm_process_monitor_msg(BMS_SRE *pp_sre, char *pp_buf)
{
    CTmTxBase *lp_tx     = NULL;
    MS_Mon_Msg  lv_msg;

    if (pp_buf == NULL)
    {
        tm_log_event(DTM_INVALID_PROC_MON_MSG, SQ_LOG_CRIT, "DTM_INVALID_PROC_MON_MSG");
        TMTrace(1, ("tm_process_monitor_msg ENTER, data null, exiting \n"));

        abort ();
    }
    
    memcpy (&lv_msg, pp_buf, sizeof (MS_Mon_Msg));

    TMTrace(2, ("tm_process_monitor_msg ENTRY, type=%d\n", lv_msg.type));

    if (lv_msg.type != MS_MsgType_NodeQuiesce)
       // Delay reply for quiesce processing.
       // At this point we will not reply with an error so get the reply
       // out of the way so we can do some real processing
       XMSG_REPLY_(pp_sre->sre_msgId,       /*msgid*/
                   NULL,           /*replyctrl*/
                   0,      /*replyctrlsize*/
                   NULL,           /*replydata*/
                   0,          /*replydatasize*/
                   0,              /*errorclass*/
                   NULL);          /*newphandle*/
    
    switch (lv_msg.type) 
    {
    case MS_MsgType_Change:
    {
         tm_process_registry_change(&lv_msg.u.change);
         break;
    }
    case MS_MsgType_Shutdown:
    {
         // If the TM is already shutting down, we don't want to change the state back to TM_STATE_SHUTTING_DOWN
         if (gv_tm_info.state_shutdown())
         {         
             TMTrace(1, ("tm_process_monitor_msg Shutdown notice, level %d.  Duplicate notice ignored!\n",
                lv_msg.u.shutdown.level));
             tm_log_event(DTM_DUPLICATE_SHUTDOWN_NOTICE, SQ_LOG_CRIT, "DTM_DUPLICATE_SHUTDOWN_NOTICE",
                FEDUP,-1,gv_tm_info.nid(),-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,lv_msg.u.shutdown.level);
         }
         else
         {
             TMTrace(1, ("tm_process_monitor_msg Shutdown notice, level %d.\n", 
                 lv_msg.u.shutdown.level));
             tm_log_event(DTM_SHUTDOWN_NOTICE, SQ_LOG_INFO, "DTM_SHUTDOWN_NOTICE",
                 -1,-1,gv_tm_info.nid(),-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,lv_msg.u.shutdown.level);

             gv_tm_info.state(TM_STATE_SHUTTING_DOWN);
             gv_tm_info.shutdown_level(lv_msg.u.shutdown.level);

             if (lv_msg.u.shutdown.level == MS_Mon_ShutdownLevel_Immediate)
                tm_abort_all_transactions(true);
         }

         // if the shutdown mode is MS_Mon_ShutdownLevel_Normal, go back to the main loop
         // to service user's request for committing or aborting the outstanding txs.
         // If the shutdown mode is MS_Mon_ShutdownLevel_Abrupt, go back to the main loop
         // and wait for the monitor to kill all the TMs.-
         break;
    }  // MS_MsgType_Shutdown
    case MS_MsgType_NodeDown:
    {
         TMTrace(3, ("tm_process_monitor_msg NodeDown notice for nid %d\n", lv_msg.u.down.nid));

         // Appoint new Lead TM if necessary.
         tm_get_leader_info();

         if (gv_tm_info.lead_tm() == false)
         {
            gv_tm_info.down_without_sync(lv_msg.u.death.nid, true);
            TMTrace(3, ("tm_process_monitor_msg - setting down_without_sync to TRUE for node %d\n",
                        lv_msg.u.death.nid))
         }
         // Process the death notice for the logical node which died
         // We may already have processed a node down message, depending on the Seaquest
         // environment - configurations with spares don't send node down.
         if (gv_tm_info.tm_is_up(lv_msg.u.death.nid))
            tm_process_node_down_msg(lv_msg.u.death.nid);

         // If we're the lead TM, attempt to recover the TM.
         if (gv_tm_info.lead_tm() == true)
            gv_tm_info.addTMRestartRetry(lv_msg.u.death.nid, 0);
         break;
    }
    case MS_MsgType_NodeUp:
    {
        TMTrace(1, ("tm_process_monitor_msg NodeUp notice for nid %d\n", lv_msg.u.up.nid));
        tm_log_event(DTM_NODEUP, SQ_LOG_INFO, "DTM_NODEUP", 
            -1,-1,gv_tm_info.nid(),-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,
           NULL,lv_msg.u.up.nid);
        break;
    }
    case MS_MsgType_NodePrepare:
    {
        TMTrace(1, ("tm_process_monitor_msg NodePrepare notice for nid %d\n", lv_msg.u.prepare.nid));
        tm_log_event(DTM_NODEPREPARE, SQ_LOG_INFO, "DTM_NODEPREPARE", 
            -1,-1,gv_tm_info.nid(),-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,
           NULL,lv_msg.u.prepare.nid);
        if (gv_tm_info.lead_tm()) {
           gv_tm_info.restart_tm_process(lv_msg.u.prepare.nid);
        }
        break;
    }
    case MS_MsgType_TmRestarted:
    {
        TMTrace(1, ("tm_process_monitor_msg TMRestarted notice for nid %d\n", lv_msg.u.tmrestarted.nid));

        // Appoint new Lead TM if necessary.
         tm_get_leader_info();
         if (gv_tm_info.lead_tm() == false)
         {
            gv_tm_info.down_without_sync(lv_msg.u.tmrestarted.nid, true);
            TMTrace(3, ("tm_process_monitor_msg - setting down_without_sync to TRUE for node %d\n",
                        lv_msg.u.tmrestarted.nid))
         }
        	
        tm_log_event(DTM_TMRESTARTED, SQ_LOG_INFO, "DTM_TMRESTARTED", 
            -1,-1,gv_tm_info.nid(),-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,
           NULL,lv_msg.u.tmrestarted.nid);
        if (gv_tm_info.lead_tm()) {
           gv_tm_info.open_restarted_tm(lv_msg.u.tmrestarted.nid);
        }
        break;
    }
    case MS_MsgType_ProcessDeath:
    {
        TMTrace(3, ("tm_process_monitor_msg Process Death notice for %s\n", 
                    lv_msg.u.death.process_name));

         switch (lv_msg.u.death.type)
         {
             case MS_ProcessType_TSE:
             {
                tm_log_event(DTM_PROCDEATH_TSE, SQ_LOG_INFO, "DTM_PROCDEATH_TSE", 
                    -1,-1,gv_tm_info.nid(),-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,lv_msg.u.death.pid,-1,-1,
                    lv_msg.u.death.process_name,lv_msg.u.death.nid);

                TMTrace(1, ("tm_process_monitor_msg death notice for TSE %s (%d, %d).\n", 
                            lv_msg.u.death.process_name, lv_msg.u.death.nid, lv_msg.u.death.pid));

                // Check to see if the TSE is still alive - this will indicate a 
                // failover rather than a crash/stop.
                int lv_nid;
                int lv_pid;
                int lv_ret = msg_mon_get_process_info ((char *) &lv_msg.u.death.process_name, 
                                                       &lv_nid, &lv_pid);

                // Mark TSE as failed in RM list
                if (lv_ret != FEOK || lv_pid == -1)
                {
                    tm_log_event(DTM_TSE_FAILURE_DETECTED, SQ_LOG_WARNING, "DTM_TSE_FAILURE_DETECTED", 
                        lv_ret,-1,gv_tm_info.nid(),-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,lv_msg.u.death.pid,-1,-1,
                        lv_msg.u.death.process_name,lv_msg.u.death.nid);
                    TMTrace(1, ("tm_process_monitor_msg Failure detected for TSE %s (%d, %d).\n", 
                            lv_msg.u.death.process_name, lv_msg.u.death.nid, lv_msg.u.death.pid));

                    gv_RMs.TSE()->fail_rm(lv_msg.u.death.nid, lv_msg.u.death.pid);
                }
                else
                {
                    // Ignore failovers, they're transparent apart from an error
                    // 201 for any outstanding I/Os.
                    RM_Info_TSEBranch * lp_RM;
                    for (int lv_inx=0; 
                         lv_inx < gv_RMs.TSE()->return_highest_index_used(); 
                         lv_inx++)
                    {
                        lp_RM = gv_RMs.TSE()->return_slot_by_index(lv_inx);
                        if (lp_RM && 
                            !strcmp(lp_RM->pname(), 
                                    (char *) &lv_msg.u.death.process_name))
                        {
                            lp_RM->nid(lv_nid);
                            lp_RM->pid(lv_pid);
                        }
                    }
                    tm_log_event(DTM_TSE_FAILOVER_DETECTED, SQ_LOG_INFO, "DTM_TSE_FAILOVER_DETECTED", 
                        -1,-1,gv_tm_info.nid(),-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,lv_pid,-1,-1,
                        lv_msg.u.death.process_name,lv_nid);
                    TMTrace(1, ("tm_process_monitor_msg failover detected for TSE %s. New primary is (%d, %d).\n", 
                            lv_msg.u.death.process_name, lv_nid, lv_pid));
                }
                break;
             }
             case MS_ProcessType_ASE:
             {
                  // Don't care unless its the TLOG.  TODO
                tm_log_event(DTM_PROCDEATH_ASE, SQ_LOG_INFO, "DTM_PROCDEATH_ASE", 
                    -1,-1,gv_tm_info.nid(),-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,lv_msg.u.death.pid,-1,-1,
                    lv_msg.u.death.process_name,lv_msg.u.death.nid);

                TMTrace(1, ("tm_process_monitor_msg death notice for ASE %s (%d, %d).\n",
                    lv_msg.u.death.process_name, lv_msg.u.death.nid, lv_msg.u.death.pid));

                break;
             }
             case MS_ProcessType_DTM:
             {
                tm_log_event(DTM_PROCDEATH_DTM, SQ_LOG_INFO, "DTM_PROCDEATH_DTM", 
                    -1,-1,gv_tm_info.nid(),-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,
                    lv_msg.u.death.process_name);
                TMTrace(1, ("tm_process_monitor_msg death notice for DTM%d\n", lv_msg.u.death.nid));
                break;
             }
             // most likely application death.  If not, then the tx will come back NULL
             // and we'll just return
             default :
             {
                 TMTrace(1, ("tm_process_monitor_msg death notice for process type %d\n",lv_msg.u.death.type ));

                  TM_Txid_Internal  *lp_transid  = (TM_Txid_Internal *)&lv_msg.u.death.transid;
                  lp_tx = (CTmTxBase *) gv_tm_info.get_tx(lp_transid);
                  if (lp_tx != NULL)
                       lp_tx->schedule_abort();
                  // this is a regular process death not associated with a transid.  Find them....
                  else 
                  {
                      TMTrace(3, ("tm_process_monitor_msg death notice for pid %d on nid %d\n",
                                        lv_msg.u.death.pid, lv_msg.u.death.nid ));

                      int64 lv_count = 0;
                      int64 lv_size = 0;
                      void **lp_tx_list = gv_tm_info.get_all_txs (&lv_size);
                      if (!lp_tx_list)
                         break;

                      TM_TX_Info *lp_current_tx = (TM_TX_Info *)lp_tx_list[lv_count];
                      while ((lv_count < lv_size) && lp_current_tx) 
                      {
                           if (lp_current_tx->is_app_partic(lv_msg.u.death.pid, lv_msg.u.death.nid))
                           {
                               TMTrace(3, ("tm_process_monitor_msg aborting seq num %d\n",
                                        lp_current_tx->seqnum() ));
                               lp_current_tx->remove_app_partic(lv_msg.u.death.pid, lv_msg.u.death.nid);
                               lp_current_tx->schedule_abort();
                           }
                           lv_count++;
                           if (lv_count < lv_size)
                              lp_current_tx = (TM_TX_Info*)lp_tx_list[lv_count];
                      }
                      if (lp_tx_list)
                          delete []lp_tx_list;
                  } 
                break;
             }
          }
       break;
    }
    case MS_MsgType_NodeQuiesce:
    {
         TMTrace(3, ("tm_process_monitor_msg NodeQuiesce notice.\n"));
         tm_process_node_quiesce_msg();
         XMSG_REPLY_(pp_sre->sre_msgId,       /*msgid*/
                     NULL,           /*replyctrl*/
                     0,      /*replyctrlsize*/
                     NULL,           /*replydata*/
                     0,          /*replydatasize*/
                     0,              /*errorclass*/
                     NULL);          /*newphandle*/
         break;
    }
    case MS_MsgType_TmSyncAbort:
    {
        // There can be many monitor replies, so circle through them all
        for (int lv_count = 0; lv_count < lv_msg.u.tmsync.count; lv_count++)
        {
            // We use sync handles for receiving DTMs and sync tags for originating DTMs.  Right now this
            // is required because the tm_sync_cb() is passed the handle.
            Tm_Sync_Data *lp_sync_data = (Tm_Sync_Data *)gv_sync_map.get(lv_msg.u.tmsync.handle[lv_count]);
         
            // originating DTM
            if (lp_sync_data == NULL)
            {
               tm_originating_sync_abort (lv_msg.u.tmsync.orig_tag[lv_count]);
            }
            // recipient DTM
            else 
            {
               gv_sync_map.remove(lv_msg.u.tmsync.handle[lv_count]);
               delete lp_sync_data;
            }
        }
        break; // case MS_MsgType_TmSyncAbort
    } 
    case MS_MsgType_TmSyncCommit:
    {
        // There can be many monitor replies, so circle through them all
        for (int lv_count = 0; lv_count < lv_msg.u.tmsync.count; lv_count++)
        {
            // We use sync handles for receiving DTMs and sync tags for originating DTMs.  Right now this
            // is required because the tm_sync_cb() is passed the handle.
            Tm_Sync_Data *lp_sync_data = (Tm_Sync_Data *)gv_sync_map.get(lv_msg.u.tmsync.handle[lv_count]);
      
            // originating DTM
            if (lp_sync_data == NULL)
            {
                tm_originating_sync_commit(lv_msg.u.tmsync.orig_tag[lv_count]);
            }
            // receipient DTM
            else
            {
                tm_recipient_sync_commit(lp_sync_data);
                gv_sync_map.remove(lv_msg.u.tmsync.handle[lv_count]);
                delete lp_sync_data;
            }
        }
    break;
    }
    case MS_MsgType_Event:
    case MS_MsgType_UnsolicitedMessage:
    default:
    {
         break;
    }
    };
   
    TMTrace(2, ("tm_process_monitor_msg EXIT\n"));
}

// -----------------------------------------------------------------------
// tm_process_msg
// Purpose - process messages incoming to the TM
// -----------------------------------------------------------------------
void tm_process_msg(BMS_SRE *pp_sre) 
{
    short                  lv_ret;
    char                   la_send_buffer[4096];
    char                   la_recv_buffer[sizeof(Tm_Req_Msg_Type)];
    char                  *la_recv_buffer_ddl = NULL;
    Tm_Broadcast_Req_Type *lp_br_req;
    Tm_Broadcast_Rsp_Type *lp_br_rsp; 
    Tm_Perf_Stats_Req_Type *lp_ps_req;
    Tm_Perf_Stats_Rsp_Type *lp_ps_rsp; 
    //Tm_Sys_Status_Req_Type *lp_ss_req;
    Tm_Sys_Status_Rsp_Type *lp_ss_rsp;
    Tm_RolloverCP_Req_Type *lp_rc_req;
    Tm_RolloverCP_Rsp_Type *lp_rc_rsp;
    Tm_Control_Point_Req_Type *lp_cp_req;
    MESSAGE_HEADER_SQ     *lp_msg_hdr;
    CTmTxMessage          *lp_msg;

    TMTrace(2, ("tm_process_msg ENTRY\n"));

    if((unsigned)(pp_sre->sre_reqDataSize) > (sizeof(Tm_Req_Msg_Type))){
       la_recv_buffer_ddl = new char[pp_sre->sre_reqDataSize];

    lv_ret = BMSG_READDATA_(pp_sre->sre_msgId,           // msgid
                            la_recv_buffer_ddl,          // reqdata
                            pp_sre->sre_reqDataSize);    // bytecount

    }else{
    lv_ret = BMSG_READDATA_(pp_sre->sre_msgId,           // msgid
                            la_recv_buffer,              // reqdata
                            pp_sre->sre_reqDataSize);    // bytecount
    }

    if (lv_ret != 0)
    {
       // a return value of 1 means the message has been abandoned by the sender.
       if (lv_ret == 1)
       {
          tm_log_event(DTM_TM_READ_MSG_FAIL, SQ_LOG_WARNING, "DTM_TM_READ_MSG_FAIL", lv_ret);
          TMTrace(1, ("tm_process_msg : BMSG_READDATA_ failed with error %d. Message ignored!\n", lv_ret));
          return;
       }
       else
       {
          tm_log_event(DTM_TM_READ_MSG_FAIL, SQ_LOG_CRIT, "DTM_TM_READ_MSG_FAIL", lv_ret);
          TMTrace(1, ("tm_process_msg : BMSG_READDATA_ failed with error %d\n", lv_ret));
           abort();
       }   
    }
    
    if (pp_sre->sre_flags & XSRE_MON) 
    {
        tm_process_monitor_msg(pp_sre, la_recv_buffer);
        return;
    }

    lp_msg_hdr = (MESSAGE_HEADER_SQ *)&la_recv_buffer;

    TMTrace(3, ("tm_process_msg : tm %d, type %d, msgid %d\n",
                    gv_tm_info.nid(), lp_msg_hdr->rr_type.request_type, pp_sre->sre_msgId));

    // Test the message version and make sure not too low OR too high
    if ((lp_msg_hdr->version.request_version < TM_SQ_MSG_VERSION_MINIMUM) ||
        (lp_msg_hdr->version.request_version > TM_SQ_MSG_VERSION_CURRENT))
    {
        tm_log_event(DTM_TM_MSG_VERSION_INVALID, SQ_LOG_CRIT, "DTM_TM_MSG_VERSION_INVALID");
        TMTrace(1, ("tm_process_msg : Old message received. Minimum supported=%d, "
                       "Received message version=%d\n",
                       TM_SQ_MSG_VERSION_MINIMUM,
                       lp_msg_hdr->version.request_version));
        // Reply with error since illegal version
        XMSG_REPLY_(pp_sre->sre_msgId,   // msgid
                NULL,                    // replyctrl 
                0,                       // replyctrlsize
                NULL,                    // replydata
                0,                       // replydatasize
                FEINCOMPATIBLEVERSION,   // errorclass
                NULL);                   // newphandle
        return;
    }

    switch (lp_msg_hdr->rr_type.request_type)
    {
   case TM_MSG_TYPE_BROADCAST:
    {
         lp_br_req = (Tm_Broadcast_Req_Type *) la_recv_buffer;
         lp_br_rsp = (Tm_Broadcast_Rsp_Type *) la_send_buffer;
         tm_initialize_rsp_hdr(lp_br_req->iv_msg_hdr.rr_type.request_type,
                               (Tm_Rsp_Msg_Type *) lp_br_rsp);
         tm_process_req_broadcast (pp_sre, lp_br_req, lp_br_rsp); 
         TMTrace(2, ("tm_process_msg EXIT\n"));
         return; 
    }
    case TM_MSG_TYPE_TMPERFSTATS:
    {
         lp_ps_req = (Tm_Perf_Stats_Req_Type *) la_recv_buffer;
         lp_ps_rsp = (Tm_Perf_Stats_Rsp_Type *) la_send_buffer;
         if (gv_tm_info.lead_tm())
         {
            // We ignore unexpected Perf Stats request because they can happen shortly after
            // a lead TM migration.
            TMTrace(1, ("tm_process_msg : Warning ignoring Performance Statistics request received by Lead TM from nid %d\n",
                    lp_ps_req->iv_sending_tm_nid));
            tm_log_event(DTM_TM_UNEXPECTED_PS_RECEIVED, SQ_LOG_WARNING, "DTM_TM_UNEXPECTED_PS_RECEIVED",
                         -1, -1, gv_tm_info.nid(), -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, 
                         -1, "Lead TM received Performance Statistics request", lp_ps_req->iv_sending_tm_nid);
         }
         tm_fill_perf_stats_buffer(lp_ps_rsp);
         ushort lv_len = sizeof(Tm_Perf_Stats_Rsp_Type);
         XMSG_REPLY_(pp_sre->sre_msgId,       // msgid
                NULL,                    // replyctrl 
                0,                       // replyctrlsize
                (char *) lp_ps_rsp,         // replydata
                lv_len,                  // replydatasize
                0,                       // errorclass
                NULL);                   // newphandle

         TMTrace(2, ("tm_process_msg EXIT\n")); 
         return; 
    }
    case TM_MSG_TYPE_CALLSTATUSSYSTEM:
    {
         //lp_ss_req = (Tm_Sys_Status_Req_Type *) la_recv_buffer;
         lp_ss_rsp = (Tm_Sys_Status_Rsp_Type *) la_send_buffer;

         TM_STATUSSYS *lp_system_status =  new TM_STATUSSYS();
         
         gv_tm_info.send_system_status(&lp_ss_rsp->iv_status_system);
         ushort lv_len = sizeof(Tm_Sys_Status_Rsp_Type);
         XMSG_REPLY_(pp_sre->sre_msgId,       // msgid
                NULL,                    // replyctrl 
                0,                       // replyctrlsize
                (char *) lp_ss_rsp,         // replydata
                lv_len,                  // replydatasize
                0,                       // errorclass
                NULL);                   // newphandle
         delete lp_system_status;

         TMTrace(2, ("tm_process_msg EXIT\n")); 
         return; 
    }
    case TM_MSG_TYPE_STATUSSYSTEM:
    {
         //lp_ss_req = (Tm_Sys_Status_Req_Type *) la_recv_buffer;
         lp_ss_rsp = (Tm_Sys_Status_Rsp_Type *) la_send_buffer;

         tm_fill_sys_status_buffer(lp_ss_rsp);
         ushort lv_len = sizeof(Tm_Perf_Stats_Rsp_Type);
         XMSG_REPLY_(pp_sre->sre_msgId,       // msgid
                NULL,                    // replyctrl 
                0,                       // replyctrlsize
                (char *) lp_ss_rsp,         // replydata
                lv_len,                  // replydatasize
                0,                       // errorclass
                NULL);                   // newphandle

         TMTrace(2, ("tm_process_msg EXIT\n")); 
         return; 
    }
    case TM_MSG_TYPE_ROLLOVER_CP:
    {
         lp_rc_req = (Tm_RolloverCP_Req_Type *) la_recv_buffer;
         lp_rc_rsp = (Tm_RolloverCP_Rsp_Type *) la_send_buffer;
         int64 lv_sequence_no = lp_rc_req->iv_sequence_no;

         TMTrace(2, ("tm_control_point_rollover nid: %d, position: %ld\n", lp_rc_req->iv_nid, lv_sequence_no)); 
     
         // May write more than one control point if lv_sequence_no == 1 and iv_audit_seqno != 1
         if((lv_sequence_no > gv_tm_info.audit_seqno()) || ((lv_sequence_no == 1) && (gv_tm_info.audit_seqno() !=1))) {
            gv_tm_info.audit_seqno(lv_sequence_no);
            gv_tm_info.addControlPointEvent();
         }

         ushort lv_len = sizeof(Tm_RolloverCP_Rsp_Type);
         XMSG_REPLY_(pp_sre->sre_msgId,  // msgid
                NULL,                    // replyctrl 
                0,                       // replyctrlsize
                (char *) lp_rc_rsp,      // replydata
                lv_len,                  // replydatasize
                0,                       // errorclass
                NULL);                   // newphandle

         TMTrace(2, ("tm_process_msg EXIT\n")); 
         return; 
    }
    case TM_MSG_TYPE_CP:
    {
          lp_cp_req = (Tm_Control_Point_Req_Type *) la_recv_buffer;
          TMTrace(3, ("tm_process_msg : Control Point from Lead TM nid %d, type %d, startup %d.\n",
                      lp_cp_req->iv_sending_tm_nid, lp_cp_req->iv_type, lp_cp_req->iv_startup));
          if (gv_tm_info.lead_tm())
          {
             // We ignore these unexpected control points because they can happen shortly after
             // a lead TM migration.
             TMTrace(1, ("tm_process_msg : Control Point request received by Lead TM from nid %d\n",
                     lp_cp_req->iv_sending_tm_nid));
             tm_log_event(DTM_TM_UNEXPECTED_CP_RECEIVED, SQ_LOG_WARNING, "DTM_TM_UNEXPECTED_CP_RECEIVED",
                          -1, -1, gv_tm_info.nid(), -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, 
                          -1, "Lead TM received Control Point request", lp_cp_req->iv_sending_tm_nid);
          }
          else
          {
             if (lp_cp_req->iv_startup)
             {
                 TMTrace(3, ("tm_process_msg : Control Point startup from Lead TM nid %d.\n",
                         lp_cp_req->iv_sending_tm_nid));
                 gv_tm_info.schedule_init_and_recover_rms();
             }
             else
             {
                 gv_system_tx_count = 0;
                 gv_tm_info.write_all_trans_state();
             }
          }
          Tm_Control_Point_Rsp_Type *lp_rsp2 = 
                        (Tm_Control_Point_Rsp_Type *) la_send_buffer;
          lp_rsp2->iv_error = 0;
          lp_rsp2->iv_msg_hdr.rr_type.reply_type = 
                       (short) (lp_msg_hdr->rr_type.request_type + 1);
          lp_rsp2->iv_msg_hdr.miv_err.error = 0;
          tm_send_reply(pp_sre->sre_msgId,  (Tm_Rsp_Msg_Type *)lp_rsp2);
          return;
    }
    case TM_MSG_TYPE_SHUTDOWN_COMPLETE:
    {
         TMTrace(3, ("tm_process_msg SHUTDOWN_COMPLETE message received.\n"));
         // This is the shutdown_complete inquiry from the lead TM.  Reply if all
         // active txs have been aborted or commited and all RMs are closed.  
         // If the lead TM fails during the Seaquest Shutdown coordination,
         // a new lead TM can take over and resend this inquiry message.
         // That's ok. Just reply to the message again.
         Tm_Shutdown_Rsp_Type *lp_rsp_shutdown = 
                               (Tm_Shutdown_Rsp_Type *) la_send_buffer;

         bool lv_shutdown = false; 
         lp_rsp_shutdown->iv_msg_hdr.rr_type.reply_type = 
                      (short) (lp_msg_hdr->rr_type.request_type + 1);

         switch (gv_tm_info.state())
         {
            case TM_STATE_WAITING_RM_OPEN:
            case TM_STATE_UP:
            case TM_STATE_SHUTTING_DOWN:
            case TM_STATE_TX_DISABLED:
            case TM_STATE_TX_DISABLED_SHUTDOWN_PHASE1:
            case TM_STATE_QUIESCE:
            case TM_STATE_DRAIN:
            {
               lp_rsp_shutdown->iv_error = FETMSHUTDOWN_NOTREADY;
               break;
            }
            case TM_STATE_SHUTDOWN_COMPLETED:
            case TM_STATE_DOWN:
            {
                TMTrace(3, ("tm_process_msg shutdown complete.\n"));
               lp_rsp_shutdown->iv_error = FEOK;
               lv_shutdown = true;
               break;
            }
            default:
            {
                TMTrace(3, ("tm_process_msg shutdown dirty.\n"));
               lp_rsp_shutdown->iv_error = FETMSHUTDOWN_FATAL_ERR;
               lv_shutdown = true;
               break;
            }
         }
         tm_send_reply(pp_sre->sre_msgId, (Tm_Rsp_Msg_Type *) lp_rsp_shutdown);

         if (lv_shutdown)
         {
              TMTrace(1, ("SHUTDOWN : Non Lead DTM%d shutting down, TM state %d.\n", gv_tm_info.nid(), gv_tm_info.state())); 
              msg_mon_process_shutdown();
              TMTrace(1, ("$TM%d exiting. TM state %d.\n",
                      gv_tm_info.nid(), gv_tm_info.state()));
            exit(0);
         }

         return;
    }
    default:
        break;
    }// switch

    // Allocate a message object.  It will be deleted by the
    // TM_TX_Info::process_eventQ method once the request
    // has been processed.

    if( la_recv_buffer_ddl!=NULL)
       lp_msg = new CTmTxMessage((Tm_Req_Msg_Type *) la_recv_buffer_ddl, pp_sre->sre_msgId, la_recv_buffer_ddl);
    else 
       lp_msg = new CTmTxMessage((Tm_Req_Msg_Type *) &la_recv_buffer, pp_sre->sre_msgId, NULL);

    if (lp_msg_hdr->dialect_type == DIALECT_TM_DP2_SQ)
    {
       tm_process_msg_from_xarm(lp_msg);
       TMTrace(2, ("tm_process_msg EXIT. XARM Request detected.\n"));
       return;
    }

   switch (lp_msg->requestType()) 
    {
    case TM_MSG_TYPE_BEGINTRANSACTION:
        tm_process_req_begin(lp_msg);
        break;
    case TM_MSG_TYPE_ENDTRANSACTION:
        tm_process_req_end(lp_msg);
        break;
    case TM_MSG_TYPE_ABORTTRANSACTION:
        tm_process_req_abort(lp_msg);
        break;
    case TM_MSG_TYPE_STATUSTRANSACTION:
        tm_process_req_status (lp_msg);
        break;
    case TM_MSG_TYPE_LISTTRANSACTION:
        tm_process_req_list (lp_msg);
        break;
    case TM_MSG_TYPE_TMSTATS:
        tm_process_req_tmstats (lp_msg);
        break;
    case TM_MSG_TYPE_STATUSTM:
        tm_process_req_statustm (lp_msg);
        break;
    case TM_MSG_TYPE_ATTACHRM:
        tm_process_req_attachrm (lp_msg);
        break;
    case TM_MSG_TYPE_STATUSTRANSMGMT:
        tm_process_req_status_transmgmt(lp_msg);
        break;
    case TM_MSG_TYPE_STATUSALLTRANSMGT:
        tm_process_req_status_all_transmgmt(lp_msg);
        break;
    case TM_MSG_TYPE_GETTRANSINFO:
        tm_process_req_status_gettransinfo(lp_msg);
        break;
    case TM_MSG_TYPE_LEADTM:
        tm_process_req_leadtm (lp_msg);
        break;
    case TM_MSG_TYPE_ENABLETRANS:
        tm_process_req_enabletrans (lp_msg);
        break;
    case TM_MSG_TYPE_DISABLETRANS:
        tm_process_req_disabletrans (lp_msg);
        break;
    case TM_MSG_TYPE_DRAINTRANS:
        tm_process_req_draintrans (lp_msg);
        break;
    case TM_MSG_TYPE_QUIESCE:
        tm_process_node_quiesce_msg(lp_msg);
        break;
    case (TM_MSG_TYPE_ENABLETRANS + TM_TM_MSG_OFFSET):
        // Non-lead TM enableTrans arriving from lead TM
        gv_tm_info.enableTrans(lp_msg);
        break;
    case (TM_MSG_TYPE_DISABLETRANS + TM_TM_MSG_OFFSET):
        // Non-lead TM disableTrans arriving from lead TM
        gv_tm_info.disableTrans(lp_msg);
        break;
    case (TM_MSG_TXINTERNAL_SHUTDOWNP1_WAIT + TM_TM_MSG_OFFSET):
        // Non-lead TM ShutdownPhase1Wait arriving from lead TM
        gv_tm_info.disableTrans(lp_msg);
        break;
    case TM_MSG_TYPE_AX_REG:
        tm_process_req_ax_reg (lp_msg);
        break;
    case TM_MSG_TYPE_JOINTRANSACTION:
        tm_process_req_join_trans (lp_msg);
        break;
    case TM_MSG_TYPE_SUSPENDTRANSACTION:
        tm_process_req_suspend_trans (lp_msg);
        break;
    case TM_MSG_TYPE_AX_UNREG:
        tm_process_req_ax_unreg (lp_msg);
        break;
    case TM_MSG_TYPE_TEST_TX_COUNT:
        lp_msg->response()->u.iv_count.iv_count = gv_tm_info.num_active_txs();
        lp_msg->reply();
        delete lp_msg;
        break;
    case TM_MSG_TYPE_DOOMTX:
        tm_process_req_doomtx(lp_msg);
        break;
    case TM_MSG_TYPE_TSE_DOOMTX:
        tm_process_req_TSE_doomtx(lp_msg);
        break;
    case TM_MSG_TYPE_WAIT_TMUP:
        tm_process_req_wait_tmup(lp_msg);
        break;
   case TM_MSG_TYPE_REGISTERREGION:
        tm_process_req_registerregion(lp_msg);
        break;
   case TM_MSG_TYPE_DDLREQUEST:
        tm_process_req_ddlrequest(lp_msg);
        break;
   case TM_MSG_TYPE_REQUESTREGIONINFO:
        tm_process_req_requestregioninfo(lp_msg);
        break;
   case TM_MSG_TYPE_GETNEXTSEQNUMBLOCK:
        tm_process_req_GetNextSeqNum(lp_msg);
        break;
   default:

        // EMS message here, DTM_INVALID_MESSAGE_TYPE
        tm_log_event(DTM_INVALID_MESSAGE_TYPE2, SQ_LOG_CRIT , "DTM_INVALID_MESSAGE_TYPE2",
                    -1, /*error_code*/ 
                    -1, /*rmid*/
                    gv_tm_info.nid(), /*dtmid*/ 
                    -1, /*seq_num*/
                    -1, /*msgid*/
                    -1, /*xa_error*/
                    -1, /*pool_size*/
                    -1, /*pool_elems*/
                    -1, /*msg_retries*/
                    -1, /*pool_high*/
                    -1, /*pool_low*/
                    -1, /*pool_max*/
                    -1, /*tx_state*/
                    lp_msg->requestType()); /*data */

        TMTrace(1, ("tm_process_msg - TM%d received UNKNOWN message type : %d\n",
               gv_tm_info.nid(), lp_msg->requestType()));
        // Reply with error since unknown request type
        XMSG_REPLY_(pp_sre->sre_msgId,   // msgid
                NULL,                    // replyctrl 
                0,                       // replyctrlsize
                NULL,                    // replydata
                0,                       // replydatasize
                FEINVALOP,               // errorclass
                NULL);                   // newphandle                
        return;
    }
    TMTrace(2, ("tm_process_msg EXIT\n"));
}

// ---------------------------------------------------------------
// tm_shutdown_helper
// Purpose - 
// ----------------------------------------------------------------
void tm_shutdown_helper ()
{
    TMTrace(2, ("tm_shutdown_helper ENTRY\n"));

    if (gv_tm_info.num_active_txs() <= 0)
    {
        TMShutdown *lp_Shutdown = new TMShutdown(&gv_tm_info, gv_RMs.TSE()->return_rms());
        gv_tm_info.shutdown_coordination_started(true);
        lp_Shutdown->coordinate_shutdown();
        delete lp_Shutdown;
        if (gv_tm_info.lead_tm())
            gv_tm_info.set_txnsvc_ready(TXNSVC_DOWN);

    }
    else
    {
        // wait 1/4 of a second, can be fine-tuned later
        XWAIT(0, TM_SHUTDOWN_WAKEUP_INTERVAL);
    }
   TMTrace(2, ("tm_shutdown_helper EXIT\n"));
}

// ---------------------------------------------------------------
// tm_main_initialize
// Purpose - call all initialization routines
// --------------------------------------------------------------
void tm_main_initialize()
{
    char            la_leader_name[BUFSIZ];
    char            la_event_data[MS_MON_MAX_SYNC_DATA];
    int32           lv_event_len;
    int32           lv_leader_nid;
    int32           lv_leader_pid;

    // initialize and get TM leader information
    gv_tm_info.initialize();
    tm_xarm_initialize();

    tm_log_event(DTM_TM_PROCESS_STARTUP, SQ_LOG_INFO, "DTM_TM_PROCESS_STARTUP", 
                 -1,-1,gv_tm_info.nid());

    /*lv_leader_error =*/ msg_mon_tm_leader_set(&lv_leader_nid,
                      &lv_leader_pid, la_leader_name);
    gv_tm_info.lead_tm_nid(lv_leader_nid);

    if (lv_leader_nid < 0  || lv_leader_nid >= MAX_NODES)
    {
        tm_log_event(DTM_TM_LEADTM_BAD, SQ_LOG_CRIT, "DTM_TM_LEADTM_BAD",
                     -1, -1, gv_tm_info.nid(), -1, -1, -1, -1, -1, -1, -1, 
                     -1, -1, -1, -1, -1, -1, NULL, lv_leader_nid);
        abort();
    }
    if (lv_leader_nid == gv_tm_info.nid())
    {
        tm_log_event (DTM_TM_LEADTM_SET, SQ_LOG_INFO , "DTM_TM_LEADTM_SET", 
                    -1, /*error_code*/ 
                    -1, /*rmid*/
                    gv_tm_info.nid(), /*dtmid*/ 
                    -1, /*seq_num*/
                    -1, /*msgid*/
                    -1, /*xa_error*/
                    -1, /*pool_size*/
                    -1, /*pool_elems*/
                    -1, /*msg_retries*/
                    -1, /*pool_high*/
                    -1, /*pool_low*/
                    -1, /*pool_max*/
                    -1, /*tx_state*/
                    -1, /*data */
                    -1, /*data1*/
                    -1,/*data2 */
                    NULL, /*string2*/
                    gv_tm_info.nid() /*node*/);
        gv_tm_info.lead_tm(true);
        
        //This must be system startup time. Wait for the events
        // before performing system recovery.
        //The AM and TSE events will be implemented in the future.
        //msg_mon_event_wait (AM_TLOG_FIXUP_COMPLETED_EVENT_ID, &lv_event_len, la_event_data);
        //msg_mon_event_wait (TSE_START_EVENT_ID, &lv_event_len, la_event_data);
        msg_mon_event_wait (DTM_START_EVENT_ID, &lv_event_len, la_event_data);
    }

     //Start timer thread
     TMTrace(1, ("tm_main_initialize, Starting timer Thread.\n")); 
     tm_start_timerThread();

     //Start example thread
     //tm_start_exampleThread();

     tm_start_auditThread();

     // Initialize the XA TM Library
     xaTM_initialize(gv_tm_info.iv_trace_level, gv_tm_info.tm_stats(), gv_tm_info.tmTimer());

     // Initialize the HBase TM Library
     HbaseTM_initialize(gv_tm_info.iv_trace_level, gv_tm_info.tm_stats(), gv_tm_info.tmTimer(), gv_tm_info.nid());

     if (gv_tm_info.lead_tm())
     {
        TMTrace(1, ("tm_main_initialize, I am Lead TM, $TM%d.\n", 
                     gv_tm_info.nid())); 
     }
    // open all tms before recovery, will return if not the lead
    gv_tm_info.open_other_tms();

    TMTrace(1, ("main : Lead DTM is on node %d\n", lv_leader_nid));

    if (gv_tm_info.lead_tm())
    {
       // init_and_recover_rms() will invoke system recovery if this
       // the Lead TM.
       gv_tm_info.schedule_init_and_recover_rms();
       gv_wait_interval = LEAD_DTM_WAKEUP_INTERVAL/10;
    }

    TMTrace(1, ("main : initialize complete, pid : %d, nid %d, cp interval %d\n", 
                     gv_tm_info.pid(), gv_tm_info.nid(), (gv_tm_info.cp_interval()/60000))); 

} 


// ----------------------------------------------------------------
// main method
// ----------------------------------------------------------------
int main(int argc, char *argv[]) 
{
    int16           lv_ret;
    int32           lv_my_nid;
    int32           lv_my_pid;
    BMS_SRE         lv_sre;

    CALL_COMP_DOVERS(tm, argc, argv);

    const int l_size = 20;
    int l_idx = 0;
    typedef struct l_element_
    {
        int16 lv_ret;
        BMS_SRE lv_sre;
    } l_element;

    l_element l_array[l_size];


    // get our pid info and initialize
    msg_init(&argc, &argv);

    // get our pid info and initialize
    msg_mon_get_my_info2(&lv_my_nid, // mon node-id
                         &lv_my_pid, // mon process-id
                         NULL,       // mon name
                         0,       // mon name-len
                         NULL,       // mon process-type
                         NULL,       // mon zone-id
                         NULL,       // os process-id
                         NULL,       // os thread-id
                         NULL);      // component-id
    gv_tm_info.nid (lv_my_nid);
    gv_tm_info.pid (lv_my_pid);
#ifdef MULTITHREADED_TM
    XWAIT(0, -2);
#endif

    msg_mon_process_startup(true); // server?
    msg_debug_hook ("tm.hook", "tm.hook");
    tm_init_logging();
    msg_mon_tmsync_register(tm_sync_cb);
    msg_mon_enable_mon_messages (1);
    msg_enable_priority_queue();
    // allow the DTM to use all the message descriptors
    XCONTROLMESSAGESYSTEM(XCTLMSGSYS_SETRECVLIMIT,XMAX_SETTABLE_RECVLIMIT);  
    XCONTROLMESSAGESYSTEM(XCTLMSGSYS_SETSENDLIMIT,SEABED_MAX_SETTABLE_SENDLIMIT_TM);
    tm_main_initialize();

    for(;;) 
    {
        int lv_msg_count = 0;

        if ((gv_tm_info.state_shutdown()) &&
           (!gv_tm_info.shutdown_coordination_started()))
        {
            tm_shutdown_helper();
        }

        XWAIT(LREQ, (int)gv_wait_interval); // 10 ms units
        do 
        {
            lv_ret = BMSG_LISTEN_((short *) &lv_sre, // sre
                                 BLISTEN_ALLOW_IREQM, 
                                 0);                   // listenertag
            l_array[l_idx].lv_ret = lv_ret;
            memcpy((void *) &l_array[l_idx].lv_sre, (void *) &lv_sre, sizeof(lv_sre));
            if (l_idx >= l_size-1)
                l_idx = 0;
            else
                l_idx++;

            if (lv_ret != BSRETYPE_NOWORK)
                 tm_process_msg(&lv_sre);

            // come up for air to allow control point processing if need be
            if (lv_msg_count++ > 100)
                 break;

        } while (lv_ret != BSRETYPE_NOWORK);
    }
}







It is not obvious to me why this change solves the problem. For example, at line 2808 below, we check pp_src->src_reqDataSize to see if it is too big to fit in la_recv_buffer, and if so, we dynamically allocate a buffer la_recv_buffer_ddl for it. It looks like the remaining code in the method makes assumptions that certain message types are always shorter than sizeof(Tm_Req_Msg_Type) though; but then the logic would read the message into the wrong buffer and it would fail in some other way. Could you provide an explanation of why it solves the problem? Also, I'm wondering if we could get rid of dynamically allocating la_recv_buffer_ddl and just use la_recv_buffer, by using dynamic array sizing as you have done here. That would simplify the logic and reduce heap pressure. @zcorrea, what do you think?