| // @@@ 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[300], 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) |
| break; |
| 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->getTableName(lv_inx); |
| 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; |
| |
| 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); |
| } |
| } |
| |
| |
| |
| |
| |
| |
