开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > 中间件编程 > 查看源码
TxManager.cxx
资源名称：blacktie-2.0.0.M3-src.zip　[点击查看]
上传用户：xfwatch
上传日期：2020-12-14
资源大小：872k
文件大小：15k
源码类别：

中间件编程

开发平台：

Java

TxManager.cxx：源码内容
  1. /*
  2.  * JBoss, Home of Professional Open Source
  3.  * Copyright 2009, Red Hat, Inc., and others contributors as indicated
  4.  * by the @authors tag. All rights reserved.
  5.  * See the copyright.txt in the distribution for a
  6.  * full listing of individual contributors.
  7.  * This copyrighted material is made available to anyone wishing to use,
  8.  * modify, copy, or redistribute it subject to the terms and conditions
  9.  * of the GNU Lesser General Public License, v. 2.1.
  10.  * This program is distributed in the hope that it will be useful, but WITHOUT A
  11.  * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
  12.  * PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more details.
  13.  * You should have received a copy of the GNU Lesser General Public License,
  14.  * v.2.1 along with this distribution; if not, write to the Free Software
  15.  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
  16.  * MA  02110-1301, USA.
  17.  */
  18. #include "ThreadLocalStorage.h"
  19. #include "XAResourceManagerFactory.h"
  20. #include "OrbManagement.h"
  21. #include "TxManager.h"
  22. #include "AtmiBrokerEnv.h"
  23. #include "ace/Thread.h"
  24. #include "txAvoid.h"
  25. #include "SynchronizableObject.h"
  27. #define TX_GUARD(cond) {
  28. FTRACE(txmlogger, "ENTER"); 
  29. if (_connection == NULL) { 
  30. LOG4CXX_DEBUG(txmlogger, (char*) "Cannot connect to an ORB"); 
  31. return TX_ERROR; 
  32. } 
  33. if (!cond) {  
  34. LOG4CXX_WARN(txmlogger, (char*) "protocol error: open: " << _isOpen << " transaction: " << getSpecific(TSS_KEY));   
  35. return TX_PROTOCOL_ERROR;   
  36. }}
  38. namespace atmibroker {
  39. namespace tx {
  41. log4cxx::LoggerPtr txmlogger(log4cxx::Logger::getLogger("TxLogManager"));
  42. TxManager *TxManager::_instance = NULL;
  43. SynchronizableObject lock_;
  45. TxManager *TxManager::get_instance()
  46. {
  47. FTRACE(txmlogger, "ENTER");
  49. lock_.lock();
  50. if (_instance == NULL)
  51. _instance = new TxManager();
  52. lock_.unlock();
  54. return _instance;
  55. }
  57. void TxManager::discard_instance()
  58. {
  59. FTRACE(txmlogger, "ENTER");
  60. if (_instance != NULL) {
  61. delete _instance;
  62. _instance = NULL;
  63. }
  64. }
  66. TxManager::TxManager() :
  67. _whenReturn(TX_COMMIT_DECISION_LOGGED), _controlMode(TX_UNCHAINED), _timeout (0l), _isOpen(false), _connection(NULL)
  68. {
  69. FTRACE(txmlogger, "ENTER");
  70. AtmiBrokerEnv::get_instance();
  71. try {
  72. _connection = ::initOrb((char*) "ots");
  73. LOG4CXX_DEBUG(txmlogger, (char*) "new CONNECTION: " << _connection);
  74. } catch (CORBA::SystemException & e) {
  75. LOG4CXX_WARN(txmlogger, (char*) "Failed to connect to ORB for TM: " << e._name() << " minor code: " << e.minor());
  76. } catch (...) {
  77. LOG4CXX_WARN(txmlogger, (char*) "Unknown error looking up ORB for TM");
  78. }
  79. }
  81. TxManager::~TxManager()
  82. {
  83. FTRACE(txmlogger, "ENTER");
  84. if (_connection) {
  85. (void) close();
  86. LOG4CXX_DEBUG(txmlogger, (char*) "deleting CONNECTION: " << _connection);
  87. shutdownBindings(_connection);
  88. AtmiBrokerEnv::discard_instance();
  89. }
  90. }
  92. CORBA::ORB_ptr TxManager::getOrb() {
  93. return _connection->orbRef;
  94. }
  96. atmibroker::tx::TxControl *TxManager::currentTx(const char *msg)
  97. {
  98. FTRACE(txmlogger, "ENTER");
  99. atmibroker::tx::TxControl *tx = NULL;
  101. if ((!_isOpen || (tx = (TxControl *) getSpecific(TSS_KEY)) == NULL || !tx->isActive(NULL, true)) && msg) {
  102. LOG4CXX_INFO(txmlogger, (char*) "protocol violation (" << msg << ") open="
  103. << _isOpen << " TSS_KEY=" << getSpecific(TSS_KEY));
  104. }
  106. return tx;
  107. }
  109. CosTransactions::Control_ptr TxManager::create_tx(TRANSACTION_TIMEOUT timeout)
  110. {
  111. CosTransactions::Control_ptr ctrl = NULL;
  113. if (!CORBA::is_nil(_txfac)) {
  114. try {
  115. ctrl = _txfac->create((long) timeout);
  116. } catch (CORBA::SystemException & e) {
  117. LOG4CXX_DEBUG(txmlogger, (char*) "Could not create tx: "
  118. << e._name() << " minor code: " << e.minor());
  119. }
  120. }
  122. return ctrl;
  123. }
  125. int TxManager::begin(void)
  126. {
  127. TX_GUARD((_isOpen && !getSpecific(TSS_KEY)));
  129. // start a new transaction
  130. TRANSACTION_TIMEOUT timeout = _timeout; // take a copy since _timeout can change at any time
  131. CosTransactions::Control_ptr ctrl = create_tx(timeout);
  133. if (CORBA::is_nil(ctrl)) {
  134. if (open_trans_factory() == TX_OK)
  135. ctrl = create_tx(timeout);
  137. if (CORBA::is_nil(ctrl)) {
  138. LOG4CXX_WARN(txmlogger, (char*) "Unable to start a new transaction (nil control)");
  140. return TX_ERROR;
  141. }
  142. }
  144. TxControl *tx = new TxControl(ctrl, (long) timeout, ACE_OS::thr_self());
  145. // associate the tx with the callers thread and enlist all open RMs with the tx
  146. int rc = TxManager::tx_resume(tx, TMNOFLAGS);
  148. if (rc != XA_OK) {
  149. // one or more RMs failed to start - roll back the transaction
  150. LOG4CXX_WARN(txmlogger, (char*) "begin: XA resume error: " << rc);
  151. try {
  152. CosTransactions::Terminator_var term = ctrl->get_terminator();
  154. if (!CORBA::is_nil(term))
  155. term->rollback();
  156. } catch (CORBA::SystemException & e) {
  157. LOG4CXX_DEBUG(txmlogger, (char*) "Could not terminate tx after tx_resume failure: "
  158. << e._name() << " minor code: " << e.minor());
  159. } catch (...) {
  160. LOG4CXX_DEBUG(txmlogger, (char*) "Could not terminate tx after tx_resume failure (generic exception)");
  161. }
  163. delete tx;
  165. return TX_ERROR;
  166. }
  168. LOG4CXX_DEBUG(txmlogger, (char*) "begin: ok");
  170. return TX_OK;
  171. }
  173. int TxManager::commit(void)
  174. {
  175. FTRACE(txmlogger, "ENTER");
  176. return complete(true);
  177. }
  179. int TxManager::rollback(void)
  180. {
  181. FTRACE(txmlogger, "ENTER");
  182. return complete(false);
  183. }
  185. int TxManager::rollback_only(void)
  186. {
  187. TxControl *tx;
  189. TX_GUARD(((tx = currentTx("rollback_only")) != NULL));
  191. // inform the local resource managers
  192. rm_end(TMFAIL);
  193. return tx->rollback_only();
  194. }
  196. int TxManager::complete(bool commit)
  197. {
  198. TxControl *tx;
  199. int outcome;
  201. TX_GUARD(((tx = currentTx("complete")) != NULL));
  203. std::map<int, int (*)(int)> &cds = tx->get_cds();
  205. if (cds.size() != 0) {
  206. LOG4CXX_WARN(txmlogger, (char*) "Ending a tx with outstanding xatmi descriptors is not allowed - rolling back");
  207. // invalidate the descriptors
  208. for (std::map<int,  int (*)(int)>::iterator i = cds.begin() ; i != cds.end(); i++) {
  209. int cd = (*i).first;
  210. int (*func)(int) = (*i).second;
  212.  LOG4CXX_DEBUG(txmlogger, (char*) "Invalidating descriptor " << cd);
  214. if (func != 0) {
  215. int rv = func(cd);
  216.   LOG4CXX_DEBUG(txmlogger, (char*) "Invalidate returned " << rv);
  217. // NOTE the invalidate function may mark the tx as rollback only
  218. }
  219. }
  221. commit = false;
  222. }
  224. // no need to call rm_end since each RM wrapper calls xa_end during the prepare call
  225. outcome = (commit ? tx->commit(reportHeuristics()) : tx->rollback());
  227. delete tx;
  229. return (isChained() ? chainTransaction(outcome) : outcome);
  230. }
  232. int TxManager::chainTransaction(int outcome)
  233. {
  234. FTRACE(txmlogger, "ENTER");
  235. /*
  236.  * NOTE: outcome will only truly represent the outcome of commit if the commit_return
  237.  * characteristic is TX_COMMIT_COMPLETED (see method reportHeuristics()).
  238.  * Using get Coordinator::get_status is ambiguous since NoTransaction can mean the
  239.  * transaction committed or rolled back and has been forgotten.
  240.  * TODO in begin register a participant in the transaction so we can definitively know
  241.  * the transaction outcome.
  242.  */
  243. switch(begin()) {
  244. case TX_OK:
  245. return TX_OK;
  246. default:
  247. switch (outcome) {
  248. case TX_OK:
  249. return TX_NO_BEGIN;
  250. case TX_ROLLBACK:
  251. return TX_ROLLBACK_NO_BEGIN;
  252. case TX_MIXED:
  253. return TX_MIXED_NO_BEGIN;
  254. case TX_HAZARD:
  255. return TX_HAZARD_NO_BEGIN;
  256. default:
  257. return outcome;
  258. }
  259. }
  260. }
  262. int TxManager::set_commit_return(COMMIT_RETURN when_return)
  263. {
  264. TX_GUARD(_isOpen);
  266. if (when_return != TX_COMMIT_DECISION_LOGGED &&
  267. when_return != TX_COMMIT_COMPLETED)
  268. return TX_EINVAL;
  270. _whenReturn = when_return;
  272. return TX_OK;
  273. }
  275. int TxManager::set_transaction_control(TRANSACTION_CONTROL mode)
  276. {
  277. TX_GUARD(_isOpen);
  279. if (mode != TX_UNCHAINED && mode != TX_CHAINED)
  280. return TX_EINVAL;
  282. _controlMode = mode;
  284. return TX_OK;
  285. }
  287. int TxManager::set_transaction_timeout(TRANSACTION_TIMEOUT timeout)
  288. {
  289. TX_GUARD(_isOpen);
  291. if (timeout < 0l)
  292. return TX_EINVAL;
  294. _timeout = timeout;
  296. return TX_OK;
  297. }
  299. int TxManager::info(void *info)
  300. {
  301. TX_GUARD(_isOpen);
  303. atmibroker::tx::TxControl *tx = currentTx(NULL);
  305.     if (info != 0) {
  306.         long whenReturn = _whenReturn;
  307.         long controlMode = _controlMode;
  308. long status = -1l;
  309.         long timeout = _timeout;
  310.         if (tx != NULL) {
  311. XAResourceManagerFactory::getXID(::getXid(info));
  312.             status = tx->get_status();
  313.         }
  314. ::updateInfo(info, whenReturn, controlMode, timeout, status);
  315.     }
  316.     LOG4CXX_DEBUG(txmlogger, (char*) "info tx=" << tx);
  317.     return (tx != NULL ? 1 : 0);
  318. }
  320. int TxManager::open_trans_factory(void)
  321. {
  322. char *transFactoryId = orbConfig.transactionFactoryName;
  324. if (transFactoryId == NULL || strlen(transFactoryId) == 0) {
  325. LOG4CXX_ERROR(txmlogger, (char*) "Please set the TRANS_FACTORY_ID env variable");
  326. return TX_ERROR;
  327. }
  329. try {
  330. CosNaming::Name *name = _connection->default_ctx->to_name(transFactoryId);
  331. LOG4CXX_DEBUG(txmlogger, (char*) "resolving Tx Fac Id: " << transFactoryId);
  332. CORBA::Object_var obj = _connection->default_ctx->resolve(*name);
  333. delete name;
  334. LOG4CXX_DEBUG(txmlogger, (char*) "resolved OK: " << (void*) obj);
  335. _txfac = CosTransactions::TransactionFactory::_narrow(obj);
  336. LOG4CXX_DEBUG(txmlogger, (char*) "narrowed OK: " << (void*) _txfac);
  337. } catch (CORBA::SystemException & e) {
  338. LOG4CXX_ERROR(txmlogger, 
  339. (char*) "Error resolving Tx Service: " << e._name() << " minor code: " << e.minor());
  340. return TX_ERROR;
  341. } catch (...) {
  342. LOG4CXX_ERROR(txmlogger, 
  343. (char*) "Unknown error resolving Tx Service did you run ant jts in the JBoss distribution and edit the jbossts properties to bind the service in the CORBA naming service: " << transFactoryId);
  344. return TX_ERROR;
  345. }
  347. return TX_OK;
  348. }
  350. int TxManager::open(void)
  351. {
  352. TX_GUARD((true));
  354. if (_isOpen)
  355. return TX_OK;
  357. if (open_trans_factory() != TX_OK)
  358. return TX_ERROR;
  360. if (rm_open() != 0) {
  361. LOG4CXX_ERROR(txmlogger, (char*) "At least one resource manager failed");
  362. (void) rm_close();
  364. return TX_ERROR;
  365. }
  367. // re-initialize values
  368. _controlMode = TX_UNCHAINED;
  369. _timeout = 0l;
  371. _isOpen = true;
  373. return TX_OK;
  374. }
  376. int TxManager::close(void)
  377. {
  378. FTRACE(txmlogger, "ENTER");
  379. if (!_isOpen)
  380. return TX_OK;
  382. TX_GUARD((getSpecific(TSS_KEY) == NULL));
  384. _isOpen = false;
  385. rm_close();
  387. return TX_OK;
  388. }
  390. int TxManager::rm_open(void)
  391. {
  392. FTRACE(txmlogger, "ENTER");
  393. try {
  394. _xaRMFac.createRMs(_connection);
  395. return 0;
  396. } catch (RMException& ex) {
  397. LOG4CXX_WARN(txmlogger, (char*) "failed to load RMs: " << ex.what());
  398. return -1;
  399. }
  400. }
  402. // private methods
  403. void TxManager::rm_close(void)
  404. {
  405. FTRACE(txmlogger, "ENTER");
  406. _xaRMFac.destroyRMs();
  407. }
  409. // pre-requisite:- there is an active transaction
  410. int TxManager::rm_end(int flags, int altflags)
  411. {
  412. FTRACE(txmlogger, "ENTER: " << std::hex << flags);
  413. TxControl *tx = (TxControl *) getSpecific(TSS_KEY);
  414. return (tx ? _xaRMFac.endRMs(tx->isOriginator(), flags, altflags) : XA_OK);
  415. }
  417. // pre-requisite:- there is an active transaction
  418. int TxManager::rm_start(int flags, int altflags)
  419. {
  420. FTRACE(txmlogger, "ENTER: " << std::hex << flags);
  421. TxControl *tx = (TxControl *) getSpecific(TSS_KEY);
  422. return (tx ? _xaRMFac.startRMs(tx->isOriginator(), flags, altflags) : XA_OK);
  423. }
  425. CosTransactions::Control_ptr TxManager::get_ots_control(long* ttl)
  426. {
  427. FTRACE(txmlogger, "ENTER");
  428. TxControl *tx = (TxControl *) getSpecific(TSS_KEY);
  430. return (tx ? tx->get_ots_control(ttl) : 0);
  431. }
  433. int TxManager::tx_resume(CosTransactions::Control_ptr control, long ttl, int flags, int altflag)
  434. {
  435. FTRACE(txmlogger, "ENTER");
  436. TxControl *tx = new TxControl(control, ttl, 0);
  437. int rc = TxManager::tx_resume(tx, flags);
  438. if (rc != XA_OK) {
  439. delete tx;
  440. }
  442. return rc;
  443. }
  445. int TxManager::tx_resume(TxControl *tx, int flags, int altflags)
  446. {
  447. FTRACE(txmlogger, "ENTER " << tx << " - flags=" << std::hex << flags);
  448. int rc = XAER_NOTA;
  450. if (getSpecific(TSS_KEY)) {
  451. TxControl *pt = (TxControl *) getSpecific(TSS_KEY);
  453. LOG4CXX_WARN(txmlogger, (char *) "Thread already bound to " << pt << " (deleting it)");
  454. delete pt;
  455. }
  457. try {
  458. // TMJOIN TMRESUME TMNOFLAGS
  459. // must associate the tx with the thread before calling start on each open RM
  460.    setSpecific(TSS_KEY, tx);
  461. if ((rc = TxManager::get_instance()->rm_start(flags, altflags)) == XA_OK) {
  462. LOG4CXX_DEBUG(txmlogger, (char *) "Resume tx: ok");
  464. return XA_OK;
  465. } else {
  466. LOG4CXX_WARN(txmlogger, (char *) "Resume tx: error: " << rc);
  467. }
  469. } catch (PortableServer::POA::ObjectAlreadyActive &e) {
  470. // THIS COMES FROM XAResourceManager.cxx:204 poa_->activate_object_with_id(objId, ra);
  471. LOG4CXX_WARN(txmlogger, (char *) "Resume tx: ObjectAlreadyActive");
  472. } catch (...) {
  473. LOG4CXX_WARN(txmlogger, (char *) "Resume tx: generic exception");
  474. }
  476. destroySpecific(TSS_KEY);
  478. return rc;
  479. }
  481. CosTransactions::Control_ptr TxManager::tx_suspend(int flags, int altflags)
  482. {
  483. FTRACE(txmlogger, "ENTER");
  484. return (tx_suspend((TxControl *) getSpecific(TSS_KEY), flags, altflags));
  485. }
  487. /**
  488.  * Suspend the transaction and return the control.
  489.  * The caller is responsible for releasing the returned control
  490.  */
  491. CosTransactions::Control_ptr TxManager::tx_suspend(TxControl *tx, int flags, int altflags)
  492. {
  493. FTRACE(txmlogger, "ENTER");
  495. if (tx && tx->isActive(NULL, true)) {
  496. // increment the control reference count
  497. CosTransactions::Control_ptr ctrl = tx->get_ots_control(NULL);
  498. // suspend all open Resource Managers (TMSUSPEND TMMIGRATE TMSUCCESS TMFAIL)
  499. (void) rm_end(flags, altflags);
  500. // disassociate the transaction from the callers thread
  501. tx->suspend();
  502. delete tx;
  504. FTRACE(txmlogger, "< ctrl: " << ctrl);
  505. return ctrl;
  506. }
  508. FTRACE(txmlogger, "< ctrl: 0x0");
  509.  return NULL;
  510. }
  512. int TxManager::resume(int cd)
  513. {
  514. FTRACE(txmlogger, "ENTER");
  515. TxControl *tx = (TxControl *) getSpecific(TSS_KEY);
  517. if (tx) {
  518. std::map<int, int (*)(int)> &cds = tx->get_cds();
  519. std::map<int,  int (*)(int)>::iterator i = cds.find(cd);
  521. if (i != cds.end()) {
  522. LOG4CXX_DEBUG(txmlogger, (char*) "Removing tp call " << cd << " from tx "
  523. << tx << " remaining tpcalls: " << cds.size());
  524. cds.erase(i);
  526. LOG4CXX_DEBUG(txmlogger, (char*) "Deleted cd - remaining tpcalls: " << cds.size());
  527. if (cds.size() == 0) {
  528. LOG4CXX_DEBUG(txmlogger, (char*) "No more outstanding calls - resume RMs");
  529. return rm_start(TMRESUME);
  530. }
  531. }
  532. }
  534. return XA_OK;
  535. }
  537. int TxManager::suspend(int cd, int (*invalidate)(int cd))
  538. {
  539. FTRACE(txmlogger, "ENTER: " << cd);
  540. TxControl *tx = (TxControl *) getSpecific(TSS_KEY);
  542. if (tx) {
  543. std::map<int, int (*)(int)> &cds = tx->get_cds();
  544. std::map<int, int (*)(int)>::iterator i = cds.find(cd);
  546. if (i == cds.end()) {
  547. LOG4CXX_DEBUG(txmlogger, (char*) "Adding tp call " << cd << " to tx " << tx);
  548. cds[cd] = invalidate;
  550. if (cds.size() == 1) {
  551. LOG4CXX_DEBUG(txmlogger, (char*) "First outstanding call - suspending RMs");
  552. return rm_end(TMSUSPEND | TMMIGRATE);
  553. }
  554. }
  555. }
  557. return XA_OK;
  558. }
  560. bool TxManager::isCdTransactional(int cd)
  561. {
  562. FTRACE(txmlogger, "ENTER: " << cd);
  563. TxControl *tx = (TxControl *) getSpecific(TSS_KEY);
  565. if (tx) {
  566. std::map<int, int (*)(int)> &cds = tx->get_cds();
  567. std::map<int, int (*)(int)>::iterator i = cds.find(cd);
  568. LOG4CXX_TRACE(txmlogger, (char*) "found=" << (i != cds.end()) << " tx=" << tx << " calls=" << cds.size());
  569. return (i != cds.end());
  570. }
  572. return false;
  573. }
  575. #if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)
  576. template class ACE_Singleton<TxManager, ACE_Null_Mutex>;
  577. #elif defined (ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA)
  578. pragma instantiate ACE_Singleton<TxManager, ACE_Null_Mutex>;
  579. #endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */
  582. } // namespace tx
  583. } //namespace atmibroker