| // @@@ 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 @@@ |
| // tmtxbase is the base class for transaction objects. This includes |
| // both transactions originating in and owned by DTM and XARM |
| // subordinate transaction branches. |
| // TM_TX_Info and CTmXaTxn are both derived from this class. |
| |
| #include <sys/types.h> |
| #include <stdlib.h> |
| #include <string.h> |
| |
| #include "seabed/ms.h" |
| #include "rm.h" |
| #include "xatmmsg.h" |
| #include "xatmlib.h" |
| #include "seabed/trace.h" |
| #include "tmlogging.h" |
| #include "tmtxbase.h" |
| #include "tminfo.h" |
| |
| // ------------------------------------- |
| // CTmTxBase Methods |
| // ------------------------------------- |
| |
| // ------------------------------------- |
| // CTmTxBase constructor |
| // Purpose : calls initialize |
| // ------------------------------------- |
| CTmTxBase::CTmTxBase(int32 pv_nid, int64 pv_flags, int32 pv_trace_level, |
| int32 pv_seq, int32 pv_pid, int32 pv_rm_wait_time) |
| //: CTmPoolElement() |
| { |
| //EID(EID_CTmTxBase); |
| // Initialize iv_trace_level and iv_lock_count here because they're used in calls |
| // to lock(). |
| iv_trace_level = pv_trace_level; |
| // Mutex attributes: Recursive = true, ErrorCheck=false |
| ip_mutex = new TM_Mutex(true, false); |
| iv_ender_nid = pv_nid; //TMs nid is the default for the ender nid. |
| ip_timeoutEvent = NULL; |
| ip_hung_event = NULL; |
| ip_Thread = NULL; |
| iv_threadPending = false; |
| ip_Qlock_owner = NULL; |
| ip_currRequest = NULL; |
| |
| memset(&iv_transid, 0, sizeof(TM_Txid_Internal)); |
| iv_tag = 0; |
| iv_tx_state = TM_TX_STATE_NOTX; |
| iv_ender_pid = 0; |
| iv_num_active_partic.set_val(0); |
| iv_prepared_rms = 0; |
| iv_cleanup_sem = 0; |
| iv_rm_wait_time = 0; |
| iv_in_use = false; |
| iv_incremented = false; |
| iv_mark_for_rollback = false; |
| iv_tm_aborted = false; |
| iv_tse_aborted = false; |
| iv_appl_aborted = false; |
| iv_heur_aborted = false; |
| iv_read_only = false; |
| iv_use_ext_transid = false; |
| iv_written_in_cp = false; |
| iv_wrote_trans_state = false; |
| iv_recovering = false; |
| iv_transactionBusy = false; |
| iv_timeout = 0; |
| ip_branches = NULL; |
| |
| // Initialize separately from TM_Info::new_tx as we don't have the creator nid,pid when called by |
| // the tmPool constructor. |
| // this->initialize(pv_nid, pv_flags, pv_trace_level, pv_seq, pv_pid, pv_rm_wait_time); |
| } |
| |
| // ------------------------------------- |
| // CTmTxBase destructor |
| // Purpose : |
| // ------------------------------------- |
| CTmTxBase::~CTmTxBase() |
| { |
| TMTrace (2, ("CTmTxBase::~CTmTxBase : ENTRY.\n")); |
| } |
| |
| |
| //---------------------------------------------------------------------------- |
| // CTmTxBase::constructPoolElement |
| // Purpose : Callback for CTmPool elements. |
| // This method must be called by the implementation class! |
| // Because the function is static we must work out here |
| // What the derived class is and call it's constructPoolElement |
| // function directly. |
| //---------------------------------------------------------------------------- |
| CTmTxBase * CTmTxBase::constructPoolElement(int64 pv_id) |
| { |
| CTmTxBase *lp_txn = NULL; |
| CTmTxKey k(pv_id); |
| TMTrace (2, ("CTmTxBase::constructPoolElement ENTRY: Constructing transaction " |
| "pool element for Txn ID (%d,%d).\n", |
| k.node(), k.seqnum())); |
| abort(); |
| /* |
| switch (iv_txnType) |
| { |
| case TM_TX_TYPE_DTM: |
| lp_txn = (CTmTxBase *) ::TM_TX_Info.constructPoolElement(pv_id); |
| break; |
| case TM_TX_TYPE_XARM: |
| lp_txn = (CTmTxBase *) ::CTmXaTxn.constructPoolElement(pv_id); |
| break; |
| default: |
| TMTrace (2, ("CTmTxBase::constructPoolElement ENTRY: Constructing transaction " |
| "pool element of type %d for Txn ID (%d,%d).\n", |
| iv_txnType, k.node(), k.seqnum())); |
| abort(); |
| } */ |
| |
| TMTrace (2, ("CTmTxBase::constructPoolElement EXIT: Transaction " |
| "pool element %p constructed.\n", lp_txn)); |
| return lp_txn; |
| } //constructPoolElement |
| |
| |
| //---------------------------------------------------------------------------- |
| // CTmTxBase::cleanPoolElement |
| // Purpose : Callback for CTmPool elements. |
| // This method is called to clean a CTmTxBase object when CTmPool::newElement |
| // allocated it from the freeList. |
| // Returns the threadNum which is reused to add the thread object to the |
| // CTmPool<CTmTxBase> inUseList. |
| //---------------------------------------------------------------------------- |
| int64 CTmTxBase::cleanPoolElement() |
| { |
| CTmTxKey k(node(), seqnum()); |
| cleanup(); |
| return k.id(); |
| } //CTmTxBase::cleanPoolElement |
| |
| |
| // --------------------------------------------------------- |
| // cleanup |
| // --------------------------------------------------------- |
| void CTmTxBase::cleanup() |
| { |
| TMTrace(2,("CTmTxBase::cleanup BASE CLASS!\n")); |
| abort(); |
| } |
| |
| |
| // --------------------------------------------------------- |
| // get_rm |
| // Purpose : Get the address of the ia_TSEBranches entry for the |
| // specified rmid. |
| // Note that this is a private function, and protected by |
| // the caller. |
| // --------------------------------------------------------- |
| RM_Info_TSEBranch * CTmTxBase::get_rm(int32 pv_rmid) |
| { |
| return branches()->TSE()->return_slot(pv_rmid); |
| } |
| |
| // ------------------------------------- |
| // initialize |
| // Purpose : Initialize this instance |
| // ------------------------------------- |
| void CTmTxBase::initialize(int32 pv_nid, int64 pv_flags, int32 pv_trace_level, |
| int32 pv_seq, int32 pv_creator_nid, int32 pv_creator_pid, |
| int32 pv_rm_wait_time) |
| { |
| TMTrace(2, ("CTmTxBase::initialize : ENTRY ID (%d,%d).\n",pv_nid, pv_seq)); |
| |
| // transid setup |
| lock(); |
| iv_txnType = TM_TX_TYPE_DTM; // default to standard DTM owned txn |
| iv_incremented = false; |
| iv_transid.iv_seq_num = pv_seq; |
| ip_Thread = NULL; |
| iv_threadPending = false; |
| iv_transid.iv_tx_flags = (short)pv_flags; //HERE |
| iv_transid.iv_node = pv_nid; |
| iv_transid.iv_version = 1; |
| iv_transid.iv_check_sum = 403; |
| iv_transid.iv_incarnation_num = gv_tm_info.incarnation_num(); |
| iv_transid.iv_tt_flags.Application = 0; |
| iv_transid.iv_tt_flags.Reserved[0] = 0; |
| iv_transid.iv_tt_flags.Reserved[1] = 0; |
| iv_transid.iv_tt_flags.Predefined = 0; |
| iv_transid.iv_timestamp = SB_Thread::Sthr::time(); |
| |
| // misc setup |
| iv_timeout = gv_tm_info.timeout(); //default timeout |
| ip_timeoutEvent = NULL; |
| iv_tm_aborted = iv_tse_aborted = iv_appl_aborted = iv_heur_aborted = iv_written_in_cp = false; |
| iv_wrote_trans_state = iv_mark_for_rollback = false; |
| iv_recovering = false; |
| iv_read_only = false; |
| iv_in_use = true; |
| iv_prepared_rms = 0; |
| iv_trace_level = pv_trace_level; |
| iv_tx_state = TM_TX_STATE_NOTX; |
| iv_tag = pv_seq; |
| iv_num_active_partic.set_val(1); |
| iv_ender_nid = pv_creator_nid; |
| iv_ender_pid = pv_creator_pid; |
| ip_currRequest = NULL; |
| iv_transactionBusy = false; |
| iv_cleanup_sem = 0; |
| iv_rm_wait_time = pv_rm_wait_time; |
| ip_hung_event = NULL; |
| add_app_partic(iv_ender_pid, iv_ender_nid); |
| |
| iv_stats.initialize(gv_tm_info.stats()->collectStats(), |
| gv_tm_info.stats()->collectInterval()); |
| iv_stats.clearCounters(); |
| |
| unlock(); |
| |
| } |
| |
| // ----------------------------------------------------------------- |
| // safe_initialize_slot |
| // purpose : this method provides a safe version of initialize_slot, to |
| // be called by external users. It acquires the lock, calls initialize_slot, |
| // releases the lock, and returns the value to the caller. |
| // ----------------------------------------------------------------- |
| int CTmTxBase::safe_initialize_slot (int32 pv_rmid) |
| { |
| int lv_idx = 0; |
| lock(); |
| lv_idx = initialize_slot (pv_rmid); |
| unlock(); |
| return lv_idx; |
| } |
| |
| // ----------------------------------------------------------------- |
| // initialize_slot |
| // purpose : this method is used to set an RM's partic flag to true |
| // so that it can partic in the transaction. |
| // Currently only called by register_branches (ax_reg), so we assume |
| // that the RM is up and set it's flags appropriately |
| // register_branches or other caller MUST have already acquired the lock. |
| // ----------------------------------------------------------------- |
| int CTmTxBase::initialize_slot (int32 pv_rmid) |
| { |
| |
| return branches()->TSE()->set_partic_and_transid(iv_transid, pv_rmid); |
| |
| } |
| |
| // ---------------------------------------------------------------- |
| // initialize_tx_rms |
| // Purpose : initialize the group of standard RMs for any |
| // given transaction. This is done at the start |
| // of every transaction |
| // ---------------------------------------------------------------- |
| void CTmTxBase::initialize_tx_rms(bool pv_partic_true) |
| { |
| lock(); |
| if (branches() == NULL) |
| { |
| TMTrace(1, ("CTmTxBase::initialize_tx_rms ERROR : ip_branches not instantiated!\n")); |
| tm_log_event(DTM_TXN_INIT_RMS_NO_BRANCH_OBJ, SQ_LOG_CRIT, "DTM_TXN_INIT_RMS_NO_BRANCH_OBJ", |
| -1,-1,node(),seqnum()); |
| abort(); |
| } |
| branches()->init_rms(this, pv_partic_true); |
| unlock(); |
| } |
| |
| |
| // -------------------------------------------------------------- |
| // sync_write |
| // Purpose - prepare and send sync for state transition |
| // ------------------------------------------------------------- |
| |
| void CTmTxBase::sync_write (int32 pv_nid, |
| int32 pv_pid, TM_TX_STATE pv_state) |
| { |
| TM_SYNC_TYPE lv_type; |
| |
| TMTrace (2, ("CTmTxBase::sync_write : ENTRY. TxnId %d, Current state %d, new state %d.\n", |
| iv_tag, iv_tx_state, pv_state)); |
| |
| iv_tx_state = pv_state; |
| |
| switch (pv_state) |
| { |
| case TM_TX_STATE_BEGINNING: |
| { |
| lv_type = TM_BEGIN_SYNC; |
| break; |
| } |
| case TM_TX_STATE_COMMITTED: |
| case TM_TX_STATE_ABORTING: |
| case TM_TX_STATE_ABORTING_PART2: |
| case TM_TX_STATE_ABORTED: |
| { |
| lv_type = TM_END_SYNC; |
| break; |
| } |
| case TM_TX_STATE_FORGOTTEN: |
| { |
| lv_type = TM_FORGET_SYNC; |
| break; |
| } |
| default: |
| { |
| return; // nothing to do for now |
| } |
| }; //switch |
| |
| init_and_send_tx_sync_data( lv_type, pv_state, |
| (TM_Transid_Type*)&iv_transid, |
| pv_nid, pv_pid); |
| |
| TMTrace (2, ("CTmTxBase::sync_write : EXIT. TxnId %d, type %d.\n", |
| iv_tag, lv_type)); |
| } |
| |
| // -------------------------------------------------------------- |
| // eventQ |
| // Purpose - retrieve the iv_eventQ to allow direct deque method |
| // calls against it. |
| // -------------------------------------------------------------- |
| TM_DEQUE * CTmTxBase::eventQ() |
| { |
| return &iv_eventQ; |
| } |
| |
| |
| // -------------------------------------------------------------- |
| // eventQ_push |
| // Purpose - push a new event to the transactions event queue. |
| // These are pushed in FIFO order. |
| // pv_threadAssociated was added to ensure that when a thread |
| // is associated with a new transaction, we don't get a race |
| // condition with the transaction being picked up by a |
| // disassociating thread before the newly instantiated/allocated |
| // thread gets a chance to process the THREAD_INITIATE event. |
| // -------------------------------------------------------------- |
| void CTmTxBase::eventQ_push(CTmTxMessage * pp_msg, bool pv_threadAssociated) |
| { |
| CTxThread * lp_thread = NULL; |
| |
| TMTrace(3, ("CTmTxBase::eventQ_push Entry: Txn ID (%d,%d), msg %d, thdAssoc %d, Thread Obj %s(%p) tid %ld, ThreadPending %d, Txn Obj %p\n", |
| node(), seqnum(), (pp_msg)?pp_msg->msgid():-1, pv_threadAssociated, |
| (ip_Thread)?ip_Thread->get_name():"MAIN", (void *) ip_Thread, |
| (ip_Thread)?ip_Thread->get_id():-1, |
| threadPending(), (void *) this)); |
| |
| if (pp_msg == NULL) |
| { |
| tm_log_event(TM_TMTX_INVALID_MSG_TYPE, SQ_LOG_CRIT, "TM_TMTX_INVALID_MSG_TYPE"); |
| TMTrace (1, ("CTmTxBase::eventQ_push - Request to be queued is NULL.\n")); |
| abort (); |
| } |
| |
| // for worker threads we need to grab a new thread for every event, unless |
| // the transaction object already has an associated thread |
| if (!pv_threadAssociated && !ip_Thread && !threadPending()) |
| { |
| TMTrace (3, ("CTmTxBase::eventQ_push : Instantiating new thread to service request.\n")); |
| threadPending(true); |
| lp_thread = gv_tm_info.new_thread(this); |
| if (lp_thread != NULL) |
| pv_threadAssociated = true; |
| else |
| threadPending(false); |
| } |
| |
| iv_eventQ.push(pp_msg); |
| |
| if(ip_Thread || threadPending()) |
| { |
| TMTrace (3, ("CTmTxBase::eventQ_push : signalling Txn ID (%d,%d) thread %s, tid %ld, event %d.\n", |
| node(), seqnum(), |
| (ip_Thread)?ip_Thread->get_name():"MAIN", |
| (ip_Thread)?ip_Thread->get_id():-1, pp_msg->requestType())); |
| iv_CV.signal(true /*lock*/); |
| } |
| else |
| { |
| if (pv_threadAssociated) |
| { |
| TMTrace (3, ("CTmTxBase::eventQ_push : " |
| "Txn ID (%d,%d), event %d queued but thread not signalled.\n", |
| node(), seqnum(), pp_msg->requestType())); |
| } |
| else |
| { |
| TMTrace (3, ("CTmTxBase::eventQ_push : no thread associated with " |
| "ID %d, event %d queued and txn added to disassociated Q.\n", |
| seqnum(), pp_msg->requestType())); |
| // Put this transaction object on the txdisassociated queue. |
| // This allows an idle thread to pick up the work. |
| // TODO: Should check here to make sure that the transaction object |
| // isn't already on the queue or we could end up with duplicates. |
| gv_tm_info.txdisassociatedQ()->push(this); |
| } |
| } |
| } |
| |
| |
| // -------------------------------------------------------------- |
| // eventQ_push_top |
| // Purpose - push an event to the top of the queue. |
| // This is used for TM internal messages where they must be |
| // processed before anything else like an Initialize. |
| // This has no effect if the TM is singlethreaded. |
| // -------------------------------------------------------------- |
| void CTmTxBase::eventQ_push_top(CTmTxMessage * pp_msg) |
| { |
| TMTrace(3, ("CTmTxBase::eventQ_push_top : Thread Obj %p(%s), lpid %ld, lname %s, Txn Obj %p, ID %d\n", |
| (void *) ip_Thread, (ip_Thread)?ip_Thread->get_name():"MAIN", |
| SB_Thread::Sthr::self_id(), SB_Thread::Sthr::self_name(), |
| (void *) this, seqnum())); |
| |
| if (pp_msg == NULL) |
| { |
| tm_log_event(TM_TMTX_INVALID_MSG_TYPE, SQ_LOG_CRIT, "TM_TMTX_INVALID_MSG_TYPE"); |
| TMTrace (1, ("CTmTxBase::eventQ_push_top - Request to be queued is NULL.\n")); |
| abort (); |
| } |
| |
| iv_eventQ.push_back(pp_msg); |
| |
| if(ip_Thread) |
| { |
| TMTrace (3, ("CTmTxBase::eventQ_push_top : signaling transaction ID %d thread %s, event %d.\n", |
| iv_tag, ip_Thread->get_name(), pp_msg->requestType())); |
| iv_CV.signal(true /*lock*/); |
| } |
| else |
| TMTrace (3, ("CTmTxBase::eventQ_push_top : no thread associated with " |
| "transaction ID %d, event %d queued.\n", |
| iv_tag, pp_msg->requestType())); |
| } |
| |
| |
| // -------------------------------------------------------------- |
| // eventQ_pop |
| // Purpose - pop an event from the end of the queue. Events are |
| // always processed in FIFO order. |
| // pp_msg must be allocated prior to calling eventQ_pop. calling |
| // the CTmTxMessage::reply method against it will delete the request |
| // so no need to clean this up. |
| // pv_timeout is only implemented as: |
| // TX_EVENTQ_WAITFOREVER (-1): wait for ever, the default, or |
| // TX_EVENTQ_NOWAIT (-2): don't wait. |
| // -------------------------------------------------------------- |
| CTmTxMessage * CTmTxBase::eventQ_pop(int pv_timeout) |
| { |
| char la_buf[1024]; |
| CTmTxMessage * lp_msg = NULL; |
| CTmTxBase *lp_txn = this; |
| long lv_thrd = SB_Thread::Sthr::self_id(); |
| long lv_txnThrd = ip_Thread->get_id(); |
| |
| TMTrace (2, ("CTmTxBase::eventQ_pop ENTRY timeout(%d) Thread %ld.\n", pv_timeout, lv_thrd)); |
| |
| if (!ip_Thread || ip_Thread->get_id() != SB_Thread::Sthr::self_id() || lv_thrd != SB_Thread::Sthr::self_id() || lv_txnThrd != lv_thrd) |
| { |
| sprintf(la_buf, "TxnThread %p, %s(%ld), Executing Thread %s(%ld), Txn Obj %p", |
| (void *) ip_Thread, |
| ((ip_Thread)?ip_Thread->get_name():"MAIN"), |
| ((ip_Thread)?ip_Thread->get_id():-1), |
| SB_Thread::Sthr::self_name(), SB_Thread::Sthr::self_id(), |
| (void *) this); |
| tm_log_event(DTM_LOGIC_ERROR_THR_MISMATCH, SQ_LOG_CRIT, "DTM_LOGIC_ERROR_THR_MISMATCH", |
| -1,-1,node(),seqnum(),-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,la_buf); |
| TMTrace (1, ("CTmTxBase::eventQ_pop : Txn ID (%d,%d) Thread Mismatch! %s\n", node(), seqnum(), la_buf)); |
| fflush(stdout); |
| abort(); |
| } |
| |
| if (pv_timeout == TX_EVENTQ_NOWAIT && iv_eventQ.empty()) |
| { |
| TMTrace (2, ("CTmTxBase::eventQ_pop EXIT nowait and queue is empty.\n")); |
| return NULL; |
| } |
| |
| while (iv_eventQ.empty()) |
| { |
| TMTrace (3, ("CTmTxBase::eventQ_pop transaction ID %d thread %s event Q " |
| "empty, waiting for signal.\n", |
| iv_tag, ((ip_Thread)?ip_Thread->get_name():"NO THREAD"))); |
| iv_CV.wait(true /*lock*/); |
| } |
| |
| lp_msg = (CTmTxMessage *) iv_eventQ.pop_end(); |
| |
| // If we've somehow picked up an event for a different thread abort now! |
| if (lp_txn != this || ip_Thread->get_id() != SB_Thread::Sthr::self_id() || lv_thrd != SB_Thread::Sthr::self_id()) |
| { |
| sprintf(la_buf, "(2) TxnThread %p, %s(%ld), Executing Thread %s(%ld), Txn Obj %p", |
| (void *) ip_Thread, |
| ((ip_Thread)?ip_Thread->get_name():"MAIN"), |
| ((ip_Thread)?ip_Thread->get_id():-1), |
| SB_Thread::Sthr::self_name(), SB_Thread::Sthr::self_id(), |
| (void *) this); |
| tm_log_event(DTM_LOGIC_ERROR_THR_MISMATCH, SQ_LOG_CRIT, "DTM_LOGIC_ERROR_THR_MISMATCH", |
| -1,-1,node(),seqnum(),-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,la_buf); |
| TMTrace (1, ("CTmTxBase::eventQ_pop : Txn ID (%d,%d) Thread Mismatch! %s\n", node(), seqnum(), la_buf)); |
| fflush(stdout); |
| abort(); |
| } |
| |
| TMTrace (2, ("CTmTxBase::eventQ_pop EXIT transaction ID %d thread %s, " |
| "msgid %d returning event %d.\n", |
| iv_tag, ((ip_Thread)?ip_Thread->get_name():"NO THREAD"), |
| ((lp_msg)?lp_msg->msgid():0), |
| ((lp_msg)?lp_msg->requestType():0))); |
| |
| return lp_msg; |
| } |
| |
| |
| // -------------------------------------------------------------- |
| // Event Specific Methods: |
| // -------------------------------------------------------------- |
| // req_begin |
| // -------------------------------------------------------------- |
| bool CTmTxBase::req_begin(CTmTxMessage * pp_msg) |
| { |
| bool lv_exit = false; |
| TMTrace (2, ("CTmTxBase::req_begin : BASE CLASS!\n")); |
| abort(); |
| return lv_exit; |
| } //req_begin |
| |
| // -------------------------------------------------------------- |
| // req_end |
| // -------------------------------------------------------------- |
| bool CTmTxBase::req_end(CTmTxMessage * pp_msg) |
| { |
| bool lv_terminate = false; |
| TMTrace (2, ("CTmTxBase::req_end : BASE CLASS!\n")); |
| abort(); |
| return lv_terminate; |
| } //req_end |
| |
| |
| // -------------------------------------------------------------- |
| // req_abort |
| // -------------------------------------------------------------- |
| bool CTmTxBase::req_abort(CTmTxMessage * pp_msg) |
| { |
| bool lv_terminateThread = false; |
| TMTrace (2, ("CTmTxBase::req_abort : BASE CLASS! Txn ID (%d,%d).\n", node(), seqnum())); |
| abort(); |
| return lv_terminateThread; |
| } //req_abort |
| |
| // -------------------------------------------------------------- |
| // rollback_txn |
| // -------------------------------------------------------------- |
| bool CTmTxBase::rollback_txn(CTmTxMessage * pp_msg) |
| { |
| TMTrace (2, ("CTmTxBase::rollback_txn : ID %d, BASE CLASS!\n", |
| seqnum())); |
| abort(); |
| return false; //Don't terminate txn thread, still need to process the |
| //TM_MSG_TXINTERNAL_ABORTCOMPLETE. |
| } //rollback_txn |
| |
| |
| // -------------------------------------------------------------- |
| // redrivecommit_txn |
| // -------------------------------------------------------------- |
| bool CTmTxBase::redrivecommit_txn(CTmTxMessage * pp_msg) |
| { |
| bool lv_terminate = false; |
| TMTrace (2, ("CTmTxBase::redrivecommit_txn BASE CLASS!: ID %d, current txn state %d\n", |
| seqnum(), tx_state())); |
| abort(); |
| return lv_terminate; |
| } //redrivecommit_txn |
| |
| // -------------------------------------------------------------- |
| // req_abort_complete |
| // -------------------------------------------------------------- |
| bool CTmTxBase::req_abort_complete(CTmTxMessage * pp_msg) |
| { |
| TMTrace (2, ("CTmTxBase::req_abort_complete BASE CLASS!: ID %d\n", seqnum())); |
| abort(); |
| return false; //Don't terminate transaction, still need to do forget |
| } |
| |
| // -------------------------------------------------------------- |
| // req_end_complete |
| // -------------------------------------------------------------- |
| bool CTmTxBase::req_end_complete(CTmTxMessage * pp_msg) |
| { |
| TMTrace (2, ("CTmTxBase::req_end_complete BASE CLASS!: ID %d\n", seqnum())); |
| abort(); |
| return false; //Don't terminate transaction, still need to do forget |
| } |
| |
| // -------------------------------------------------------------- |
| // req_forget |
| // -------------------------------------------------------------- |
| bool CTmTxBase::req_forget(CTmTxMessage * pp_msg) |
| { |
| bool lv_terminate = true; // Normal processing is for forget to |
| // drive thread disassociation for |
| // transaction threads. |
| TMTrace (2, ("CTmTxBase::req_forget BASE CLASS!: ID %d\n", seqnum())); |
| abort(); |
| return lv_terminate; //Finished with the transaction, disassociate and cleanup |
| } // req_forget |
| |
| // -------------------------------------------------------------- |
| // req_begin_complete |
| // -------------------------------------------------------------- |
| bool CTmTxBase::req_begin_complete(CTmTxMessage * pp_msg) |
| { |
| TMTrace (2, ("CTmTxBase::req_begin_complete BASE CLASS!: ID %d, error=%d\n", |
| seqnum(), pp_msg->responseError())); |
| abort(); |
| return false; //Leave transaction thread alive |
| } |
| |
| // -------------------------------------------------------------- |
| // Event Specific Methods implemented in base class |
| // -------------------------------------------------------------- |
| // req_join |
| // Purpose - Txn specific processing for JOINTRANSACTION. |
| // -------------------------------------------------------------- |
| bool CTmTxBase::req_join(CTmTxMessage * pp_msg) |
| { |
| TMTrace(2, ("CTmTxBase::req_join : ENTRY: XA Txn ID (%d,%d).\n", |
| node(), seqnum())); |
| |
| if (tx_state() != TM_TX_STATE_NOTX && |
| tx_state() != TM_TX_STATE_BEGINNING && |
| tx_state() != TM_TX_STATE_ACTIVE) |
| { |
| TMTrace (1, ("CTmTxBase::req_join : FEENDEDTRANSID\n")); |
| pp_msg->responseError(mapErr(FEENDEDTRANSID)); |
| } |
| else |
| { |
| // Transaction is either BEGINNING or ACTIVE, so can be joined. |
| // Note that a transaction in NOTX state has not yet reached BEGINNING. |
| TMTrace (3, ("CTmTxBase::req_join : active participants %d\n", |
| num_active_partic())); |
| pp_msg->responseError(FEOK); |
| if (pp_msg->request()->u.iv_join_trans.iv_coord_role == true) |
| { |
| if (iv_ender_pid == -1) |
| { |
| iv_ender_pid = pp_msg->request()->u.iv_join_trans.iv_pid; |
| inc_active_partic(); |
| add_app_partic(iv_ender_pid, pp_msg->request()->u.iv_join_trans.iv_nid); |
| } |
| else |
| pp_msg->responseError(mapErr(FETXNOTSUSPENDED)); |
| } |
| else |
| { |
| inc_active_partic(); |
| add_app_partic(pp_msg->request()->u.iv_join_trans.iv_pid, |
| pp_msg->request()->u.iv_join_trans.iv_nid); |
| } |
| } |
| |
| |
| pp_msg->response()->u.iv_join_trans.iv_tm_pid = gv_tm_info.nid(); |
| pp_msg->reply(mapErr(FEOK)); |
| |
| TMTrace (2, ("CTmTxBase::req_join : EXIT, replying error %d\n", |
| pp_msg->responseError())); |
| |
| // Return code doesn't matter, we run from the main thread |
| return false; |
| } //req_join |
| |
| // -------------------------------------------------------------- |
| // req_status |
| // Purpose - Txn specific processing for STATUSTRANSACTION. |
| // We return a status of Aborted if the transaction is in the |
| // process of aborting because a TMF_DOOMTRANSACTION_ will |
| // leave the transaction marked for rollback once an ENDTRANSACTION |
| // or ABORTTRANSACTION is received. SQL use STATUSTRANSACTION |
| // to determine whether the transaction was doomed. |
| // -------------------------------------------------------------- |
| bool CTmTxBase::req_status(CTmTxMessage * pp_msg) |
| { |
| TMTrace (2, ("CTmTxBase::req_status : ENTRY\n")); |
| |
| // For transactions that are beginning we want to report them as active |
| if (tx_state() == TM_TX_STATE_BEGINNING || |
| tx_state() == TM_TX_STATE_NOTX) |
| pp_msg->response()->u.iv_status_trans.iv_status = TM_TX_STATE_ACTIVE; |
| else |
| { |
| if (isAborting() && tx_state() != TM_TX_STATE_ABORTING && |
| tx_state() != TM_TX_STATE_ABORTING_PART2 && |
| tx_state() != TM_TX_STATE_ACTIVE) |
| pp_msg->response()->u.iv_status_trans.iv_status = TM_TX_STATE_ABORTED; |
| else |
| pp_msg->response()->u.iv_status_trans.iv_status = (short) tx_state(); |
| } |
| |
| pp_msg->reply(mapErr(FEOK)); |
| |
| TMTrace (2, ("CTmTxBase::req_status : EXIT, state %d\n", tx_state())); |
| |
| // Return code doesn't matter, we run from the main thread |
| return false; |
| } //req_status |
| |
| // -------------------------------------------------------------- |
| // req_suspend |
| // Purpose - Txn specific processing for SUSPENDTRANSACTION. |
| // -------------------------------------------------------------- |
| bool CTmTxBase::req_suspend(CTmTxMessage * pp_msg) |
| { |
| TMTrace (2, ("CTmTxBase::req_suspend : ENTRY\n")); |
| |
| if (tx_state() == TM_TX_STATE_NOTX) |
| { |
| TMTrace (1, ("CTmTxBase::req_suspend : FENOTRANSID\n")); |
| pp_msg->responseError(mapErr(FENOTRANSID)); |
| } |
| else |
| { |
| if (remove_app_partic(pp_msg->request()->u.iv_suspend_trans.iv_pid, |
| pp_msg->request()->u.iv_suspend_trans.iv_nid)) |
| { |
| dec_active_partic(); |
| |
| TMTrace (3, ("CTmTxBase::req_suspend : active participants %d\n", |
| num_active_partic())); |
| if ((pp_msg->request()->u.iv_suspend_trans.iv_coord_role == false) |
| && (pp_msg->request()->u.iv_suspend_trans.iv_pid == iv_ender_pid)) |
| iv_ender_pid = -1; // someone better call join with coord = true |
| pp_msg->responseError(mapErr(FEOK)); |
| } |
| else |
| { |
| TMTrace (3, ("CTmTxBase::req_suspend : not an active participant.\n")); |
| pp_msg->responseError(mapErr(FENOTRANSID)); |
| } |
| } |
| pp_msg->reply(); |
| |
| TMTrace (2, ("CTmTxBase::req_suspend : EXIT\n")); |
| |
| // Return code doesn't matter, we run from the main thread |
| return false; |
| } //req_suspend |
| |
| // -------------------------------------------------------------- |
| // req_ax_reg |
| // Purpose - Txn Thread specific processing for ax_reg. |
| // -------------------------------------------------------------- |
| bool CTmTxBase::req_ax_reg(CTmTxMessage * pp_msg) |
| { |
| TMTrace (2, ("CTmTxBase::req_ax_reg : ID (%d,%d) ENTRY for rmid %d\n", |
| node(), seqnum(), pp_msg->request()->u.iv_ax_reg.iv_rmid)); |
| |
| short lv_error = register_branch(pp_msg->request()->u.iv_ax_reg.iv_rmid, pp_msg); |
| |
| 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(); |
| pp_msg->reply(mapErr(lv_error)); |
| |
| // Don't reset here - we're now called in the main thread rather than queued to the |
| // transaction thread |
| // We're finished so allow other work to commence |
| //reset_transactionBusy(); |
| |
| TMTrace (2, ("CTmTxBase::req_ax_reg : ID %d, EXIT\n", seqnum())); |
| // ax_reg never causes the thread to terminate |
| return false; |
| } //req_ax_reg |
| |
| // -------------------------------------------------------------- |
| // req_registerRegion |
| // Purpose - Txn Thread specific processing for registerRegion. |
| // This is only received in SeaTrans for HBase-trx participants. |
| // -------------------------------------------------------------- |
| bool CTmTxBase::req_registerRegion(CTmTxMessage * pp_msg) |
| { |
| TMTrace (2, ("CTmTxBase::req_registerRegion : ID (%d,%d) ENTRY.\n", |
| node(), seqnum())); |
| |
| short lv_error = branches()->registerRegion(this, 0, pp_msg); |
| |
| // Reply here if we haven't yet done so. |
| lock(); |
| if (pp_msg->replyPending()) |
| pp_msg->reply(lv_error); |
| unlock(); |
| |
| reset_transactionBusy(); |
| |
| TMTrace (2, ("CTmTxBase::req_registerRegion : EXIT error %d, Txn ID (%d,%d).\n", |
| lv_error, node(), seqnum())); |
| |
| return false; //Never terminate the thread if there's no more work to do |
| } //req_registerRegion |
| |
| |
| //------------------------------------------------------------------------------ |
| // register_branch |
| // Purpose - Process received ax_reg from TSE |
| // If pp_msg == NULL, we are registering a branch internally (eg because we |
| // just received a doomtxn from a TSE which was not participating. |
| // Returns error code. |
| //----------------------------------------------------------------------------- |
| short CTmTxBase::register_branch(int32 pv_rmid, CTmTxMessage * pp_msg = NULL) |
| { |
| short lv_error = FEOK; |
| bool lv_txnActive = false, |
| lv_TTflagsChanged = false, |
| lv_IownLock = false, |
| lv_partic = false; |
| RM_Info_TSEBranch *lp_rm_branch = NULL; |
| TMTrace (2, ("CTmTxBase::register_branch ENTER, Txn ID (%d,%d), rmid %d, Tx state %d.\n", |
| node(), seqnum(), pv_rmid, tx_state())); |
| |
| // The transaction object may have the lock, but we |
| // still want to accept the ax_reg here because failure |
| // to do so could hang the TSE which won't respond to |
| // requests like xa_rollback until it gets the reply to |
| // ax_reg! |
| branches()->branch_lock(); |
| if (ip_mutex->lock_count() == 0) |
| { |
| lv_IownLock = true; |
| lock(); |
| } |
| |
| // Get the branch |
| // This had better work or our pv_rmid is bad. |
| lp_rm_branch = branches()->TSE()->return_slot(pv_rmid); |
| if (lp_rm_branch == 0) |
| { |
| tm_log_event(DTM_TMTX_INVALID_RMID, SQ_LOG_WARNING, "DTM_TMTX_INVALID_RMID", |
| FEINVTRANSID,pv_rmid,node(),seqnum()); |
| TMTrace(1, ("CTmTxBase::register_branch - Unable to find RM %d to " |
| "register branch for txn ID (%d,%d), transaction rejected with FEINVTRANSID(75).\n", |
| pv_rmid, node(), seqnum())); |
| // Reject the register request |
| return FEINVTRANSID; |
| } |
| |
| // Allow branches to participate in transactions in the following states: |
| if (iv_tx_state == TM_TX_STATE_ACTIVE || |
| iv_tx_state == TM_TX_STATE_IDLE || |
| iv_tx_state == TM_TX_STATE_ABORTING || |
| iv_tx_state == TM_TX_STATE_COMMITTING || |
| iv_tx_state == TM_TX_STATE_PREPARING || |
| iv_tx_state == TM_TX_STATE_PREPARED || |
| iv_tx_state == TM_TX_STATE_HUNGABORTED || |
| iv_tx_state == TM_TX_STATE_ABORTING_PART2) |
| { |
| lv_txnActive = true; |
| lv_partic = lp_rm_branch->add_partic(iv_transid); |
| if (lv_partic) |
| { |
| if (iv_tx_state != TM_TX_STATE_ACTIVE && |
| iv_tx_state != TM_TX_STATE_IDLE) { |
| TMTrace(1, ("CTmTxBase::register_branch - adding late participant rmid %d to Txn ID (%d,%d), Tx state %d.\n", |
| pv_rmid, node(), seqnum(), tx_state())); |
| } |
| else { |
| TMTrace(1, ("CTmTxBase::register_branch - adding participant rmid %d to Txn ID (%d,%d), Tx state %d.\n", |
| pv_rmid, node(), seqnum(), tx_state())); |
| } |
| // Issue a warning if this was a doomtxn |
| if (!pp_msg) |
| { |
| tm_log_event(DTM_TMTX_DOOMTXN_ADDED_PARTIC, SQ_LOG_WARNING, "DTM_TMTX_DOOMTXN_ADDED_PARTIC", |
| -1, pv_rmid, node(), seqnum(),-1,-1,-1,-1,-1,-1,-1,-1,tx_state()); |
| TMTrace(3, ("CTmTxBase::register_branch - RM %d already participating in Txn " |
| "ID (%d,%d), state %d.\n", |
| pv_rmid, node(), seqnum(), tx_state())); |
| } |
| branches()->TSE()->inc_num_rm_partic(); |
| } |
| else //add_partic() returns false if the branch is already participating. |
| { |
| // If we're already resolved this branch, return an error |
| lv_error = FEINVTRANSID; |
| // Only log an event if this register request came from an ax_reg. Doomtxn |
| // doesn't pass in a message pointer. |
| if (pp_msg) |
| tm_log_event(DTM_TMTX_ALREADY_PARTIC, SQ_LOG_WARNING, "DTM_TMTX_ALREADY_PARTIC", |
| lv_error, pv_rmid, node(), seqnum(),-1,-1,-1,-1,-1,-1,-1,-1,tx_state()); |
| TMTrace(3, ("CTmTxBase::register_branch - RM %d already participating in Txn " |
| "ID (%d,%d), state %d, returning error %d.\n", |
| pv_rmid, node(), seqnum(), tx_state(), lv_error)); |
| } |
| TMTrace(3, ("CTmTxBase::register_branch - RM %d ID (%d,%d) participating: %d, resolved: %d\n", |
| pv_rmid, node(), seqnum(),lv_partic, lp_rm_branch->resolved() )); |
| |
| } |
| if (pp_msg) |
| { |
| memcpy(&pp_msg->response()->u.iv_ax_reg.iv_xid, &lp_rm_branch->iv_xid, sizeof(XID)); |
| // We need to add in any flags set by the TSE that we don't know about. |
| // If the TSE set new flags then we need to write a trans state record |
| // now so that we know the new flag settings in the case of a failure. |
| int64 lv_TTflags = TT_flags() | pp_msg->request()->u.iv_ax_reg.iv_flags; |
| if (lv_TTflags != TT_flags()) |
| { |
| lv_TTflagsChanged = true; |
| TMTrace(3, ("CTmTxBase::register_branch : Txn ID (%d,%d) Detected " |
| "different TT flags from TSE RM %d, TM " |
| PFLLX ", TSE " PFLLX ", combined " PFLLX ".\n", |
| node(), seqnum(), pv_rmid, TT_flags(), |
| pp_msg->request()->u.iv_ax_reg.iv_flags, lv_TTflags)); |
| char lv_TTflagsStr[10], *lp_TTflagsStr = (char *) &lv_TTflagsStr; |
| sprintf(lp_TTflagsStr, "0x" PFLLX, lv_TTflags); |
| tm_log_event(DTM_TMTX_REG_TT_FLAGS_CHANGED, SQ_LOG_INFO, "DTM_TMTX_REG_TT_FLAGS_CHANGED", |
| -1,pv_rmid,node(),seqnum(),-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,lp_TTflagsStr); |
| // Record that a NO UNDO occured. |
| if (TT_flags() & TM_TT_NO_UNDO) |
| { |
| tm_log_event(DTM_NOUNDO, SQ_LOG_WARNING, "DTM_NOUNDO", |
| -1,pv_rmid,node(),seqnum(),-1,-1,-1,-1,-1,-1,-1,-1,tx_state()); |
| TMTrace(1, ("CTmTxBase::register_branch : WARNING - Txn ID (%d,%d), state %d - RM %d set " |
| "NO UNDO. This will leave the database in an inconsistent state.\n", |
| node(), seqnum(), tx_state(), pv_rmid)); |
| } |
| TT_flags(lv_TTflags); |
| gv_tm_info.write_trans_state (&iv_transid, iv_tx_state, abort_flags(), false); |
| } |
| pp_msg->response()->u.iv_ax_reg.iv_flags = TT_flags(); |
| } |
| if (!lv_txnActive && !lv_TTflagsChanged) |
| { |
| tm_log_event(DTM_TMTX_REG_ERROR_TOOLATE, SQ_LOG_WARNING, "DTM_TMTX_REG_ERROR_TOOLATE", |
| -1,pv_rmid,node(),seqnum(),-1,-1,-1,-1,-1,-1,-1,-1,tx_state(), |
| lp_rm_branch->resolved(),lp_rm_branch->partic()); |
| TMTrace(1, ("CTmTxBase::register_branch - ERROR RM %d is too late to participate in " |
| "Txn ID (%d,%d), state %d, partic %d, resolved %d.\n", |
| pv_rmid, node(), seqnum(), tx_state(), |
| lp_rm_branch->partic(), lp_rm_branch->resolved())); |
| lv_error = FEINVTRANSID; |
| } |
| |
| if (pp_msg) |
| pp_msg->responseError(lv_error); |
| |
| branches()->branch_unlock(); |
| |
| if (lv_IownLock) |
| unlock(); |
| |
| TMTrace(2, ("register_branch EXIT, error %d\n", lv_error)); |
| return lv_error; |
| } // CTmTxBase::register_branch |
| |
| |
| // ---------------------------------------------------------------- |
| // internal_abortTrans |
| // Purpose - Queue a rollback request for this transaction. |
| // This is currently only used during takeover, shutdown and |
| // process death. |
| // ---------------------------------------------------------------- |
| int32 CTmTxBase::internal_abortTrans(bool pv_takeover_or_shutdown) |
| { |
| CTmTxMessage * lp_msg = new CTmTxMessage(TM_MSG_TXINTERNAL_ROLLBACK); |
| |
| lp_msg->request()->u.iv_rollback_internal.iv_takeover_or_shutdown = pv_takeover_or_shutdown; |
| |
| queueToTransaction(&iv_transid, lp_msg); |
| |
| TMTrace (2, ("CTmTxBase::internal_abortTrans : EXIT\n")); |
| |
| return FEOK; |
| } //internal_abortTrans |
| |
| |
| // ---------------------------------------------------------------- |
| // redrive_rollback |
| // Purpose - Queue a redrive rollback request for this transaction. |
| // This is currently only used during a takeover. |
| // ---------------------------------------------------------------- |
| int32 CTmTxBase::redrive_rollback() |
| { |
| CTmTxMessage * lp_msg = new CTmTxMessage(TM_MSG_TXINTERNAL_REDRIVEROLLBACK); |
| |
| queueToTransaction(&iv_transid, lp_msg); |
| |
| TMTrace (2, ("CTmTxBase::redrive_rollback : EXIT\n")); |
| |
| return FEOK; |
| } //redrive_rollback |
| |
| // ---------------------------------------------------------------- |
| // redrive_commit |
| // Purpose - Queue a redrive commit request for this transaction. |
| // This is currently only used during a takeover. |
| // ---------------------------------------------------------------- |
| int32 CTmTxBase::redrive_commit() |
| { |
| CTmTxMessage * lp_msg = new CTmTxMessage(TM_MSG_TXINTERNAL_REDRIVECOMMIT); |
| |
| queueToTransaction(&iv_transid, lp_msg); |
| |
| TMTrace (2, ("CTmTxBase::redrive_commit : EXIT\n")); |
| |
| return FEOK; |
| } //redrive_commit |
| |
| |
| void CTmTxBase::schedule_redrive_sync() |
| { |
| |
| CTmTxMessage * lp_msg = new CTmTxMessage(TM_MSG_TXINTERNAL_REDRIVESYNC); |
| eventQ_push(lp_msg); |
| } |
| |
| |
| //----------------------------------------------------------------------------- |
| // decrement & test cleanup semaphore |
| // Only return true of the semaphore is 1 to make sure |
| // that only one caller will do the actual cleanup. |
| // The idea is that every thread working on a specific transaction calls |
| // inc_cleanup() when they want the txn object cleaned up on exit. |
| // The cleanup routine executes when the thread disassociates, calls |
| // dec_cleanup() which will only return true when the iv_cleanup_sem count |
| // is 1, so only one thread will cleanup the txn object. |
| // Note that we lock the Qmutex when incrementing the cleanup |
| // semaphore to ensure no one trys to write to the eventQ or PendingRequestQ |
| // while we're setting cleanup. |
| //----------------------------------------------------------------------------- |
| bool CTmTxBase::dec_cleanup() |
| { |
| bool cleanup = false; |
| |
| lock(); |
| if (iv_cleanup_sem == 1) |
| cleanup = true; |
| if (iv_cleanup_sem > 0) |
| iv_cleanup_sem--; |
| unlock(); |
| |
| TMTrace (2, ("CTmTxBase::dec_cleanup: semaphore = %d, returning %d\n", iv_cleanup_sem, cleanup)); |
| return cleanup; |
| } |
| // increment cleanup semaphore |
| void CTmTxBase::inc_cleanup() |
| { |
| lock(); |
| Qlock(); |
| iv_cleanup_sem++; |
| Qunlock(); |
| unlock(); |
| |
| TMTrace (2, ("CTmTxBase::inc_cleanup: semaphore = %d\n", iv_cleanup_sem)); |
| } |
| |
| //----------------------------------------------------------------------------- |
| // set_transactionBusy |
| // The following 2 methods allow setting and resetting the |
| // iv_transactionBusy within the same lock to ensure |
| // it's consistent and another thread doesn't interfere. |
| // Qlock() must be called before calling set_transactionBusy and Qunlock() |
| // after any outstanding request is queued to eventQ or PendingRequestQ. |
| // The return value indicates whether set_transactionBusy succeeded in setting |
| // the value. |
| // We also set the iv_currRequest here. It may be needed later to handle |
| // a request which spans waits. |
| // Returns true if iv_transactionBusy==false, ie if the transaction object |
| // is available. |
| // false if iv_transactionBusy==true - the transaction object is busy. |
| //----------------------------------------------------------------------------- |
| bool CTmTxBase::set_transactionBusy(CTmTxMessage * pp_msg) |
| { |
| bool lv_return = false; |
| |
| TMTrace (2, ("CTmTxBase::set_transactionBusy : ENTRY ID (%d,%d), iv_transactionBusy=%d.\n", |
| node(), seqnum(), iv_transactionBusy)); |
| |
| if (iv_transactionBusy == false) |
| { |
| iv_transactionBusy = true; |
| lv_return = true; |
| // Save the request, it will be needed to drive events after a wakeup for |
| // multi-part requests like ENDTRANSACTION. |
| currRequest(pp_msg); |
| } |
| else |
| { |
| TMTrace (3, ("CTmTxBase::set_transactionBusy : WARNING: Possible " |
| "programming bug. iv_transactionBusy already set on call " |
| "to set_transactionBusy.\n")); |
| if (thread() == NULL && currRequest() == NULL) |
| { |
| TMTrace (3, ("CTmTxBase::set_transactionBusy : BUG! Transaction " |
| "ID (%d,%d) busy but has no associated thread and current request!\n", |
| node(), seqnum())); |
| abort(); |
| } |
| } |
| |
| TMTrace (2, ("CTmTxBase::set_transactionBusy : EXIT ID (%d,%d), returning " |
| "transactionBusy=%d.\n", node(), seqnum(), lv_return)); |
| |
| return lv_return; |
| } //set_transactionBusy |
| |
| |
| //----------------------------------------------------------------------------- |
| // reset_transactionBusy |
| // Resets the transactionBusy flag. This method also checks the |
| // PendingRequest queue and pops the top entry if there is one. |
| // Note that we also delete the iv_currRequest message object here, |
| // so callers had better have replied to the client prior to calling |
| // reset_transactionBusy! |
| // This routine locks the Qmutex to serialize changes to the transactionBusy |
| // flag and writes/reads to the PendingRequestQ. It does not lock the transaction |
| // object. To ensure we don't mess up, |
| // pv_cleanupPending input. |
| // True if cleaning up the transaction object. This happens in forget(). |
| // In this case we cleanup the PendingRequestQ, replying to any that have |
| // outstanding replies. |
| // Returns true if iv_transactionBusy==true, ie if the transaction object |
| // was busy. |
| // false if iv_transactionBusy==false - the transaction object was free. |
| //----------------------------------------------------------------------------- |
| bool CTmTxBase::reset_transactionBusy(bool pv_cleanupPending) |
| { |
| CTmTxMessage *lp_msg = NULL; |
| CTmTxMessage *lp_firstMsg = NULL; |
| |
| TMTrace (2, ("CTmTxBase::reset_transactionBusy : ENTRY, ID (%d,%d), " |
| "cleanupPending=%d, iv_transactionBusy=%d.\n", |
| node(), seqnum(), pv_cleanupPending, iv_transactionBusy)); |
| |
| Qlock(); |
| if (iv_transactionBusy) |
| { |
| // Clean up the old message, we've finished with it |
| delete_currRequest(); |
| |
| // Check the PendingRequest queue |
| lp_msg = lp_firstMsg = (CTmTxMessage *) iv_PendingRequestQ.pop_end(); |
| while (lp_msg) |
| { |
| TMTrace (3, ("CTmTxBase::reset_transactionBusy : API " |
| "request popped off PendingRequestQ for msgid(%d).\n", |
| lp_msg->msgid())); |
| // If the cleanupPending flag was set then reply |
| // to all outstanding requests. |
| if (pv_cleanupPending) |
| { |
| if (lp_msg->replyPending()) |
| { |
| if (isAborting()) |
| lp_msg->reply(mapErr(FEABORTEDTRANSID)); |
| else |
| lp_msg->reply(mapErr(FEENDEDTRANSID)); |
| } |
| delete lp_msg; |
| lp_msg = (CTmTxMessage *) iv_PendingRequestQ.pop_end(); |
| } |
| else |
| { |
| eventQ_push(lp_msg); |
| // Make this new request the current one for the transaction |
| currRequest(lp_msg); |
| lp_msg = NULL; // Break out of while |
| } |
| } //while |
| |
| // Reset the transactionBusy flag now if there's nothing more to process |
| if (currRequest() == NULL) |
| { |
| iv_transactionBusy = false; |
| TMTrace (3, ("CTmTxBase::reset_transactionBusy: No requests outstanding, " |
| "iv_transactionBusy=false.\n")); |
| } |
| } |
| Qunlock(); |
| |
| TMTrace (2, ("CTmTxBase::reset_transactionBusy : EXIT ID (%d,%d), returning " |
| "transactionBusy=%d.\n", node(), seqnum(), iv_transactionBusy)); |
| |
| return iv_transactionBusy; |
| } //reset_transactionBusy |
| |
| |
| // --------------------------------------------------------- |
| // txn_object_good |
| // Purpose : Check that the transaction object is allocated |
| // to this transaction. There is a window in which the |
| // transaction might have been cleaned up and the transaction |
| // object returned to the free list. |
| // |
| // 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 CTmTxBase::txn_object_good(TM_Txid_Internal *pp_transid, CTmTxMessage * pp_msg) |
| { |
| if (!cleaning_up() && iv_in_use && iv_transid.iv_seq_num == pp_transid->iv_seq_num) |
| { |
| TMTrace(2, ("CTmTxBase::txn_object_good txn object is good :-)\n")); |
| return true; |
| } |
| else |
| { |
| TMTrace(1, ("CTmTxBase::txn_object_good txn %d is old! Replying " |
| "FEINVTRANSID to late request %d & deleting msg object.\n", |
| pp_transid->iv_seq_num, pp_msg->requestType())); |
| pp_msg->reply(mapErr(FEINVTRANSID)); |
| delete pp_msg; |
| return false; |
| } |
| } |
| |
| |
| // -------------------------------------------------------------- |
| // CTmTxBase::reply_to_queuedRequests |
| // Purpose : reply to all requests queued on the iv_PendingRequestQ |
| // with the error provided. |
| // -------------------------------------------------------------- |
| |
| void CTmTxBase::reply_to_queuedRequests(short pv_error) |
| { |
| CTmTxMessage * lp_msg; |
| |
| TMTrace (2, ("CTmTxBase::reply_to_queuedRequests ENTRY\n")); |
| |
| lock(); |
| // Reply to any events queued against the transaction |
| lp_msg = (CTmTxMessage *) eventQ_pop(TX_EVENTQ_NOWAIT); |
| while (lp_msg) |
| { |
| if (lp_msg->replyPending()) |
| { |
| TMTrace (3, ("CTmTxBase::reply_to_queuedRequests replying to queued " |
| "transaction event %d, msgid %d\n", |
| lp_msg->requestType(), lp_msg->msgid())); |
| lp_msg->reply(mapErr(pv_error)); |
| } |
| else |
| TMTrace (3, ("CTmTxBase::reply_to_queuedRequests queued transaction " |
| "event %d, msgid %d ignored, no reply pending.\n", |
| lp_msg->requestType(), lp_msg->msgid())); |
| |
| // If this message matches the current Request then clear it as we've just |
| // processed the message |
| if (lp_msg == ip_currRequest) |
| ip_currRequest = 0; |
| delete lp_msg; |
| lp_msg = (CTmTxMessage *) eventQ_pop(TX_EVENTQ_NOWAIT); |
| } |
| |
| // Reply to pending requests for this transaction. These can't be handled |
| // because we're in the process of forgetting the transaction. |
| lp_msg = (CTmTxMessage *) iv_PendingRequestQ.pop_end(); |
| while (lp_msg) |
| { |
| TMTrace (3, ("CTmTxBase::reply_to_queuedRequests replying to request %d\n", |
| lp_msg->requestType())); |
| |
| lp_msg->reply(mapErr(pv_error)); |
| delete lp_msg; |
| lp_msg = (CTmTxMessage *) iv_PendingRequestQ.pop_end(); |
| } |
| |
| // Check to see if we have a request queued that needs to be replied to. |
| if (ip_currRequest && ip_currRequest->replyPending()) |
| { |
| TMTrace (3, ("CTmTxBase::reply_to_queuedRequests replying to queued event %d thread %s, txn tag %d.\n", |
| ip_currRequest->requestType(), |
| ((ip_Thread)?ip_Thread->get_name():"NO THREAD"), |
| iv_tag)); |
| |
| ip_currRequest->reply(mapErr(pv_error)); |
| delete_currRequest(); |
| } |
| unlock(); |
| |
| TMTrace (2, ("CTmTxBase::reply_to_queuedRequests EXIT\n")); |
| } //reply_to_queuedRequests |
| |
| |
| // --------------------------------------------------------------------------- |
| // CTmTxBase::add_app_partic |
| // Purpose : add the process to the list of processes participating in this |
| // transaction. |
| // pv_pid pid of participating process |
| // pv_nid nid of participating process - not of this TM!!! |
| // Returns true if successfully added. |
| // false indicates we didn't get the memory for the pid, or it's |
| // already in the list! |
| // --------------------------------------------------------------------------- |
| bool CTmTxBase::add_app_partic (int32 pv_pid, int32 pv_nid) |
| { |
| CTmTxKey lv_key(pv_nid, pv_pid); |
| |
| CTmTxKey *lp_key = (CTmTxKey *) iv_app_partic_list.get(lv_key.id()); |
| if (lp_key == NULL) |
| { |
| lp_key = new CTmTxKey(lv_key.id()); |
| if (lp_key) |
| { |
| iv_app_partic_list.put(lv_key.id(), lp_key); |
| TMTrace (3, ("CTmTxBase::add_app_partic added process (%d,%d) : %p\n", |
| pv_nid, pv_pid, (void *)lp_key)); |
| return true; |
| } |
| else |
| { |
| TMTrace(1, ("CTmTxBase::add_app_partic (%d,%d) no memory to add int32 pid!, aborting.\n", |
| pv_nid, pv_pid)); |
| tm_log_event(DTM_TM_ADD_APP_PARTIC_NOMEM, SQ_LOG_CRIT, "DTM_TM_ADD_APP_PARTIC_NOMEM", |
| -1,-1,pv_nid,pv_pid); |
| abort(); |
| } |
| } |
| else |
| { |
| TMTrace (1, ("CTmTxBase::add_app_partic Warning: nid %d, pid %d found in " |
| "app_partic_list - ignored.\n", pv_nid, pv_pid)); |
| } |
| return false; |
| } //add_app_partic |
| |
| |
| bool CTmTxBase::remove_app_partic (int32 pv_pid, int32 pv_nid) |
| { |
| CTmTxKey lv_key(pv_nid, pv_pid); |
| |
| CTmTxKey *lp_key = (CTmTxKey *) iv_app_partic_list.remove(lv_key.id()); |
| if (lp_key) |
| { |
| TMTrace (3, ("CTmTxBase::remove_app_partic removing process (%d,%d) : %p\n", |
| pv_nid, pv_pid, (void *)lp_key)); |
| delete lp_key; |
| return true; |
| } |
| return false; |
| } |
| |
| // erase_app_partic is similar to remove_app_partic but can be called by |
| // code which reads through the map using get_first, ... |
| // Pass in the map element pointer. Note that in this case the contents |
| // of the map pointer is the key. |
| bool CTmTxBase::erase_app_partic(CTmTxKey *pp_key) |
| { |
| if (pp_key->id()) |
| { |
| TMTrace (3, ("CTmTxBase::erase_app_partic removing process (%d,%d) : %p\n", |
| pp_key->node(), pp_key->pid(), (void *) pp_key)); |
| iv_app_partic_list.erase(pp_key->id()); |
| delete pp_key; |
| return true; |
| } |
| return false; |
| } |
| |
| bool CTmTxBase::is_app_partic (int32 pv_pid, int32 pv_nid) |
| { |
| CTmTxKey lv_key(pv_nid, pv_pid); |
| |
| CTmTxKey *lp_key = (CTmTxKey *)iv_app_partic_list.get(lv_key.id()); |
| if (lp_key) |
| return true; |
| else |
| return false; |
| } |
| |
| void CTmTxBase::cleanup_app_partic() |
| { |
| // go through our particpation list and make sure there is no |
| // outstanding particpants (this is fine as it could have been |
| // a unilateral abort or shutdown). Just get rid of it. |
| int32 lv_count = 0; |
| CTmTxKey *lp_key = NULL; |
| CTmTxKey *lp_last_key = NULL; |
| |
| lp_key = (CTmTxKey *) iv_app_partic_list.get_first(); |
| while (lv_count < num_active_partic() && lp_key) |
| { |
| lv_count++; |
| lp_last_key = lp_key; |
| lp_key = (CTmTxKey *) iv_app_partic_list.get_next(); |
| erase_app_partic(lp_last_key); |
| } |
| iv_app_partic_list.get_end(); |
| |
| if (lv_count != num_active_partic()) |
| TMTrace(1, ("CTmTxBase::cleanup_app_partic Warning: actual participants " |
| "%d does not match stored count %d\n", |
| lv_count, num_active_partic())); |
| num_active_partic(); |
| } |
| |
| |
| //---------------------------------------------------------------------------- |
| // PendingRequestQ_push |
| // Purpose : Push a message onto the iv_PendingRequestQ. This method also |
| // checks whether the CTmTxBase object is associated with a thread. If it |
| // is not, then we add the transaction to the txdisassociatedQ to make sure |
| // it is woken when a thread becomes available. |
| //---------------------------------------------------------------------------- |
| void CTmTxBase::PendingRequestQ_push(CTmTxMessage *pp_msg) |
| { |
| TMTrace (2, ("CTmTxBase::PendingRequestQ_push ENTRY. ID %d, msgid %d.\n", |
| tag(), pp_msg->msgid())); |
| |
| PendingRequestQ()->push(pp_msg); |
| |
| |
| // If there is no thread associated here we must add the txn object to the txdisassociatedQ |
| // to ensure it's picked up when a thread becomes available. |
| if (thread() == NULL) |
| { |
| gv_tm_info.txdisassociatedQ()->push(this); |
| if (iv_trace_level >= 3) |
| trace_printf ("CTmTxBase::PendingRequestQ_push ID %d, msgid %d. Txn object " |
| "added to txdisassociatedQ. " PFLL " entries on txdisassociatedQ.\n", |
| tag(), pp_msg->msgid(), gv_tm_info.txdisassociatedQ()->size()); |
| } |
| |
| TMTrace (2, ("CTmTxBase::PendingRequestQ_push EXIT. ID %d, msgid %d. " |
| "" PFLL " entries on PendingRequest queue.\n", |
| tag(), pp_msg->msgid(), PendingRequestQ()->size())); |
| } //CTmTxBase::PendingRequestQ_push |
| |
| |
| |
| // ---------------------------------------------------------------------------- |
| // CTmTxBase::queueToTransaction |
| // Purpose : Queues the request against the transaction object. |
| // Check that we aren't already processing a request which alters the transaction state |
| // (Begin, Abort, or End). If we are, then put this request on the PendingRequestQ. |
| // If the transaction is not processing a state altering request, then queue the request |
| // directly to the transactions event queue. |
| // Returns true if it successfully queued the message against the transaction |
| // object. Otherwise it returns false. |
| // ---------------------------------------------------------------------------- |
| bool CTmTxBase::queueToTransaction(TM_Txid_Internal *pp_transid, CTmTxMessage *pp_msg) |
| { |
| bool lv_success = false; |
| |
| TMTrace (2, ("CTmTxBase::queueToTransaction : ENTRY\n")); |
| |
| // There is not need to lock the transaction object here since we're only writing to the |
| // event queue. |
| |
| Qlock(); |
| if (txn_object_good(pp_transid, pp_msg)) |
| { |
| TMTrace (3, ("CTmTxBase::queueToTransaction txn ID (%d,%d), " |
| "Request %d queued directly to txn object event queue for msgid(%d), entries in eventQ %ld.\n", |
| node(), seqnum(), pp_msg->requestType(), pp_msg->msgid(), iv_eventQ.size())); |
| eventQ_push(pp_msg, (ip_Thread?true:false)); |
| lv_success = true; |
| } |
| Qunlock(); |
| |
| TMTrace (2, ("CTmTxBase::queueToTransaction : EXIT returning %d, entries in eventQ %ld\n", |
| lv_success, iv_eventQ.size())); |
| |
| return lv_success; |
| } // CTmTxBase::queueToTransaction |
| |
| |
| // ---------------------------------------------------------------------------- |
| // CTmTxBase::currRequest |
| // Purpose : set the current request. |
| // ---------------------------------------------------------------------------- |
| void CTmTxBase::currRequest(CTmTxMessage * pp_msg) |
| { |
| if (ip_currRequest != NULL) |
| { |
| tm_log_event(DTM_TMTX_INVALID_STATE_CURR_REQ, SQ_LOG_CRIT, "DTM_TMTX_INVALID_STATE_CURR_REQ", |
| -1, /*error_code*/ |
| -1, /*rmid*/ |
| -1, /*dtmid*/ |
| iv_transid.iv_seq_num, /*seq_num*/ |
| ip_currRequest->msgid(), /*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,/*string1*/ |
| -1, /* node =*/ |
| pp_msg->msgid() /*msgid2*/ |
| ); |
| TMTrace (1, ("CTmTxBase::currRequest - Failed to set currRequest for Transaction ID %d, current " |
| "msgid %d, new msgid %d\n", |
| iv_transid.iv_seq_num, ip_currRequest->msgid(), pp_msg->msgid())); |
| abort(); |
| } |
| |
| ip_currRequest = pp_msg; |
| } // CTmTxBase::currRequest |
| |
| |
| // ---------------------------------------------------------------------------- |
| // CTmTxBase::addTimerEvent |
| // Purpose : Wrapper to simplify the addition of timer events to the timer |
| // threads event queue by transaction. Events will always be queued back |
| // against the calling transaction objects event queue. |
| // pv_type is a message type |
| // pv_delayInterval is the interval the timer thread will wait before |
| // posting the event back to the transaction in msec. |
| // ---------------------------------------------------------------------------- |
| CTmTimerEvent * CTmTxBase::addTimerEvent(TM_MSG_TYPE pv_type, int pv_delayInterval) |
| { |
| TMTrace (2, ("CTmTxBase::addTimerEvent : ENTRY, msg type %d, delay %d\n", |
| pv_type, pv_delayInterval)); |
| |
| CTmTimerEvent *lp_timerEvent = new CTmTimerEvent(pv_type, thread(), this, pv_delayInterval); |
| |
| gv_tm_info.tmTimer()->eventQ_push((CTmEvent *) lp_timerEvent); |
| |
| return lp_timerEvent; |
| } // CTmTxBase::addTimerEvent |
| |
| |
| // ---------------------------------------------------------------------------- |
| // CTmTxBase::setAbortTimeout |
| // Purpose : Set the abort timeout for this transaction. This involves |
| // creating a new timer event for the timeout. |
| // pv_timeout is the timeout in seconds. |
| // -1 = wait forever. |
| // 0 = use default (gv_tm_info.timeout()) |
| // ---------------------------------------------------------------------------- |
| void CTmTxBase::setAbortTimeout(int32 pv_timeout) |
| { |
| int32 lv_timeout; |
| int32 lv_msec; |
| |
| if (pv_timeout != 0) |
| lv_timeout = pv_timeout; |
| else |
| lv_timeout = gv_tm_info.timeout(); |
| lv_msec = (lv_timeout==-1)?-1:(lv_timeout*1000); |
| TMTrace (2, ("CTmTxBase::setAbortTimeout : ENTRY, timeout %d\n", |
| lv_timeout)); |
| |
| if (ip_timeoutEvent != NULL) |
| gv_xaTM.tmTimer()->cancelEvent(ip_timeoutEvent); |
| |
| if (lv_timeout != -1) |
| ip_timeoutEvent = addTimerEvent(TM_MSG_TXINTERNAL_ROLLBACK, lv_msec); |
| } // CTmTxBase::setAbortTimeout |
| |
| |
| // ---------------------------------------------------------------------------- |
| // CTmTxBase::addHungTimerEvent |
| // Purpose : Add a hung timer event for this transaction. Only one of these |
| // is allowed per transaction at any time. |
| // pv_type should be TM_MSG_TXINTERNAL_REDRIVEROLLBACK or TM_MSG_TXINTERNAL_REDRIVECOMMIT. |
| // ---------------------------------------------------------------------------- |
| void CTmTxBase::addHungTimerEvent(TM_MSG_TYPE pv_type) |
| { |
| TMTrace (2, ("CTmTxBase::addHungTimerEvent : ENTRY ID %d, msg type %d\n", |
| seqnum(), pv_type)); |
| if (ip_hung_event == 0) |
| ip_hung_event = addTimerEvent(pv_type, |
| gv_tm_info.trans_hung_retry_interval()); |
| } //addHungTimerEvent |
| |
| |
| // ---------------------------------------------------------------------------- |
| // CTmTxBase::hung_redrive |
| // Redrive a hung transaction. |
| // This routine walks through the global RM info table |
| // ---------------------------------------------------------------------------- |
| void CTmTxBase::hung_redrive() |
| { |
| //RM_Info_TSEBranch *lp_rmInfo; // Txn copy of RM info. |
| |
| TMTrace (2, ("CTmTxBase::hung_redrive : ENTRY ID %d\n", seqnum())); |
| TM_MSG_TYPE lv_type; |
| switch (tx_state()) |
| { |
| case TM_TX_STATE_HUNGABORTED: |
| lv_type = TM_MSG_TXINTERNAL_REDRIVEROLLBACK; |
| break; |
| case TM_TX_STATE_HUNGCOMMITTED: |
| lv_type = TM_MSG_TXINTERNAL_REDRIVECOMMIT; |
| break; |
| default: |
| // Ignore, not hung. |
| return; |
| } // switch state |
| |
| /*Commented out as I don't think we mark the RMs as failed over or recovering |
| in the ia_TSEBranches table. |
| for (int lv_index = 0; ((lv_index <= ia_TSEBranches.return_highest_index_used()) && |
| (lv_index < lv_count)); lv_index++) |
| { |
| lp_rmInfo = ia_TSEBranches.get_slot(lv_index); |
| if (lp_rmInfo->state() == TSEBranch_FAILOVER) |
| { |
| lp_rmInfo->state(TSEBranch_UP); |
| // Set the RM's 'partic' flag to true |
| safe_initialize_slot(lp_rmInfo->rmid()); |
| inc_prepared_rms(); |
| } |
| } // for each rm |
| */ |
| |
| CTmTxMessage * lp_msg = new CTmTxMessage(lv_type); |
| queueToTransaction(&iv_transid, lp_msg); |
| |
| TMTrace (2, ("CTmTxBase::hung_redrive : EXIT\n")); |
| } //CTmTxBase::hung_redrive |
| |
| |
| // ------------------------------------------------------------ |
| // CTmTxBase::associate_tx |
| // Instantiate a transaction thread if required and associate |
| // it with this transaction. |
| // Queue the specified event to this transaction object. |
| // Returns true if a thread has been associated with this |
| // transaction. |
| // false if no thread was associated. |
| // ------------------------------------------------------------ |
| bool CTmTxBase::associate_tx(CTmTxMessage *pp_msg) |
| { |
| TMTrace (2, ("CTmTxBase::associate_tx : ENTRY Txn ID (%d,%d) request %d.\n", |
| node(), seqnum(), pp_msg->requestType())); |
| |
| // Transaction threads instantiate a thread object |
| // and associate it with the transaction for its whole life. |
| bool lv_threadAssociated = false; |
| if (gv_tm_info.threadModel() == transaction) |
| { |
| TMTrace(3, ("CTmTxBase::associate_tx : Instantiating new thread.\n")); |
| CTxThread * lp_thread = gv_tm_info.new_thread(this); |
| if (lp_thread != NULL) |
| lv_threadAssociated = true; |
| } |
| |
| TMTrace(2, ("CTmTxBase::associate_tx : EXIT returning threadAssociated %d.\n", |
| lv_threadAssociated)); |
| return lv_threadAssociated; |
| } //associate_tx |
| |
| |
| // ------------------------------------------------------------ |
| // CTmTxBase::recover_tx |
| // This is used during recovery to instantiate and associate |
| // a thread with this transaction, and queue a request against |
| // it. |
| // ------------------------------------------------------------ |
| void CTmTxBase::recover_tx(TM_MSG_TYPE pv_type) |
| { |
| TMTrace (2, ("CTmTxBase::recover_tx : ENTRY Txn ID (%d,%d) request %d.\n", |
| node(), seqnum(), pv_type)); |
| CTmTxMessage * lp_msg = new CTmTxMessage(pv_type); |
| bool lv_threadAssociated = associate_tx(lp_msg); |
| |
| stats()->txnTotal()->start(); |
| // Set the appropriate state before queuing the request |
| switch (pv_type) |
| { |
| case TM_MSG_TXINTERNAL_REDRIVEROLLBACK: |
| stats()->txnAbort()->start(); |
| tx_state(TM_TX_STATE_ABORTING); |
| break; |
| case TM_MSG_TXINTERNAL_REDRIVECOMMIT: |
| stats()->txnCommit()->start(); |
| tx_state(TM_TX_STATE_COMMITTING); |
| break; |
| default: |
| // Should never get here |
| TMTrace(1, ("CTmTxBase::recover_tx : PROGRAMMING ERROR!\n")); |
| tm_log_event(DTM_RECOV_BAD_TYPE, SQ_LOG_CRIT, "DTM_RECOV_BAD_TYPE"); |
| abort(); |
| } |
| |
| // Setting the transactionBusy must succeed here because no one |
| // else knows about the transaction yet! |
| Qlock(); |
| bool lv_txnFree = set_transactionBusy(lp_msg); |
| Qunlock(); //No need to wait here, we will only queue against eventQ |
| if (!lv_txnFree) |
| { |
| TMTrace(1, ("CTmTxBase::recover_tx : ERROR transaction collision - txn ID (%d,%d) busy!\n", |
| node(), seqnum())); |
| tm_log_event(DTM_RECOV_TRANSACTION_COLLISION, SQ_LOG_CRIT, "DTM_RECOV_TRANSACTION_COLLISION", |
| -1, -1, node(), seqnum()); |
| abort(); |
| } |
| // Queue the begin against the transaction object |
| eventQ_push(lp_msg, lv_threadAssociated); |
| |
| TMTrace(2, ("CTmTxBase::recover_tx : EXIT.\n")); |
| } //recover_tx |
| |
| |
| // -------------------------------------------------------------- |
| // process_eventQ |
| // -------------------------------------------------------------- |
| void CTmTxBase::process_eventQ() |
| { |
| TMTrace (2, ("CTmTxBase::process_eventQ : BASE CLASS! ID %d, thread %s.\n", |
| iv_tag, ((gv_tm_info.multithreaded())?ip_Thread->get_name():"SINGLE THREAD"))); |
| abort(); |
| } |
| |
| |
| // ------------------------------------------------------------- |
| // schedule_eventQ |
| // ------------------------------------------------------------- |
| void CTmTxBase::schedule_eventQ() |
| { |
| TMTrace (2, ("CTmTxBase::schedule_eventQ BASE CLASS! ID %d, state %d.\n", |
| iv_tag, iv_tx_state)); |
| abort(); |
| } |
| |
| |
| // -------------------------------------------------------------- |
| // schedule_abort |
| // -------------------------------------------------------------- |
| int32 CTmTxBase::schedule_abort() |
| { |
| int32 lv_error = FEOK; |
| TMTrace (2, ("CTmTxBase::schedule_abort BASE CLASS!, Txn ID (%d,%d), state %d.\n", |
| node(), seqnum(), iv_tx_state)); |
| abort(); |
| return lv_error; |
| } |
| |
| |
| // -------------------------------------------------------------- |
| // doom_txn |
| // -------------------------------------------------------------- |
| short CTmTxBase::doom_txn() |
| { |
| short lv_error = FEOK; |
| TMTrace (2, ("CTmTxBase::doom_txn BASE CLASS!, ID %d, state %d.\n", |
| seqnum(), iv_tx_state)); |
| abort(); |
| return lv_error; |
| } |
| |
| |
| // -------------------------------------------------------------- |
| // redriverollback_txn |
| // -------------------------------------------------------------- |
| bool CTmTxBase::redriverollback_txn(CTmTxMessage * pp_msg) |
| { |
| bool lv_terminate = false; |
| TMTrace (2, ("CTmTxBase::redriverollback_txn BASE CLASS!: ID %d.\n", seqnum())); |
| abort(); |
| return lv_terminate; |
| } |
| |
