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

源码开发语言/平台
当前位置：首页 > 源码/资料 > 数据库系统 > MySQL数据库 > 查看源码
NdbEventOperationImpl.cpp
资源名称：mysql-4.1.16-win-src.zip　[点击查看]
上传用户：romrleung
上传日期：2022-05-23
资源大小：18897k
文件大小：30k
源码类别：

MySQL数据库

开发平台：

Visual C++

NdbEventOperationImpl.cpp：源码内容
  1. /* Copyright (C) 2003 MySQL AB
  3.    This program is free software; you can redistribute it and/or modify
  4.    it under the terms of the GNU General Public License as published by
  5.    the Free Software Foundation; either version 2 of the License, or
  6.    (at your option) any later version.
  8.    This program is distributed in the hope that it will be useful,
  9.    but WITHOUT ANY WARRANTY; without even the implied warranty of
  10.    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11.    GNU General Public License for more details.
  13.    You should have received a copy of the GNU General Public License
  14.    along with this program; if not, write to the Free Software
  15.    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
  18. #include <ndb_global.h>
  19. #include <kernel_types.h>
  21. #include "NdbDictionaryImpl.hpp"
  22. #include "API.hpp"
  23. #include <NdbOut.hpp>
  24. #include "NdbApiSignal.hpp"
  25. #include "TransporterFacade.hpp"
  26. #include <signaldata/CreateEvnt.hpp>
  27. #include <signaldata/SumaImpl.hpp>
  28. #include <SimpleProperties.hpp>
  29. #include <Bitmask.hpp>
  30. #include <AttributeHeader.hpp>
  31. #include <AttributeList.hpp>
  32. #include <NdbError.hpp>
  33. #include <BaseString.hpp>
  34. #include <UtilBuffer.hpp>
  35. #include <NdbDictionary.hpp>
  36. #include <Ndb.hpp>
  37. #include "NdbImpl.hpp"
  38. #include "DictCache.hpp"
  39. #include <portlib/NdbMem.h>
  40. #include <NdbRecAttr.hpp>
  41. #include <NdbEventOperation.hpp>
  42. #include "NdbEventOperationImpl.hpp"
  44. /*
  45.  * Class NdbEventOperationImpl
  46.  *
  47.  *
  48.  */
  50. //#define EVENT_DEBUG
  53. NdbEventOperationImpl::NdbEventOperationImpl(NdbEventOperation &N,
  54.      Ndb *theNdb, 
  55.      const char* eventName, 
  56.      const int bufferLength) 
  57.   : NdbEventOperation(*this), m_ndb(theNdb),
  58.     m_state(ERROR), m_bufferL(bufferLength)
  59. {
  61.   m_eventId = 0;
  62.   theFirstRecAttrs[0] = NULL;
  63.   theCurrentRecAttrs[0] = NULL;
  64.   theFirstRecAttrs[1] = NULL;
  65.   theCurrentRecAttrs[1] = NULL;
  66.   sdata = NULL;
  67.   ptr[0].p = NULL;
  68.   ptr[1].p = NULL;
  69.   ptr[2].p = NULL;
  71.   // we should lookup id in Dictionary, TODO
  72.   // also make sure we only have one listener on each event
  74.   if (!m_ndb) { ndbout_c("m_ndb=NULL"); return; }
  76.   NdbDictionary::Dictionary *myDict = m_ndb->getDictionary();
  77.   if (!myDict) { ndbout_c("getDictionary=NULL"); return; }
  79.   const NdbDictionary::Event *myEvnt = myDict->getEvent(eventName);
  80.   if (!myEvnt) { ndbout_c("getEvent()=NULL"); return; }
  82.   m_eventImpl = &myEvnt->m_impl;
  83.   if (!m_eventImpl) { ndbout_c("m_impl=NULL"); return; }
  85.   m_bufferHandle = m_ndb->getGlobalEventBufferHandle();
  86.   if (m_bufferHandle->m_bufferL > 0) 
  87.     m_bufferL =m_bufferHandle->m_bufferL;
  88.   else
  89.     m_bufferHandle->m_bufferL = m_bufferL;
  91.   m_state = CREATED;
  92. }
  94. NdbEventOperationImpl::~NdbEventOperationImpl()
  95. {
  96.   int i;
  97.   if (sdata) NdbMem_Free(sdata);
  98.   for (i=0 ; i<3; i++) {
  99.     if (ptr[i].p) NdbMem_Free(ptr[i].p);
  100.   }
  101.   for (i=0 ; i<2; i++) {
  102.     NdbRecAttr *p = theFirstRecAttrs[i];
  103.     while (p) {
  104.       NdbRecAttr *p_next = p->next();
  105.       m_ndb->releaseRecAttr(p);
  106.       p = p_next;
  107.     }
  108.   }
  109.   if (m_state == NdbEventOperation::EXECUTING) {
  110.     stop();
  111.     // m_bufferHandle->dropSubscribeEvent(m_bufferId);
  112.     ; // We should send stop signal here
  113.   }
  114. }
  116. NdbEventOperation::State
  117. NdbEventOperationImpl::getState()
  118. {
  119.   return m_state;
  120. }
  122. NdbRecAttr*
  123. NdbEventOperationImpl::getValue(const char *colName, char *aValue, int n)
  124. {
  125.   if (m_state != NdbEventOperation::CREATED) {
  126.     ndbout_c("NdbEventOperationImpl::getValue may only be called between instantiation and execute()");
  127.     return NULL;
  128.   }
  130.   NdbColumnImpl *tAttrInfo = m_eventImpl->m_tableImpl->getColumn(colName);
  132.   if (tAttrInfo == NULL) {
  133.     ndbout_c("NdbEventOperationImpl::getValue attribute %s not found",colName);
  134.     return NULL;
  135.   }
  137.   return NdbEventOperationImpl::getValue(tAttrInfo, aValue, n);
  138. }
  140. NdbRecAttr*
  141. NdbEventOperationImpl::getValue(const NdbColumnImpl *tAttrInfo, char *aValue, int n)
  142. {
  143.   // Insert Attribute Id into ATTRINFO part. 
  144.   NdbRecAttr *&theFirstRecAttr = theFirstRecAttrs[n];
  145.   NdbRecAttr *&theCurrentRecAttr = theCurrentRecAttrs[n];
  146.       
  147.   /************************************************************************
  148.    * Get a Receive Attribute object and link it into the operation object.
  149.    ************************************************************************/
  150.   NdbRecAttr *tRecAttr = m_ndb->getRecAttr();
  151.   if (tRecAttr == NULL) { 
  152.     exit(-1);
  153.     //setErrorCodeAbort(4000);
  154.     return NULL;
  155.   }
  157.   /**********************************************************************
  158.    * Now set the attribute identity and the pointer to the data in 
  159.    * the RecAttr object
  160.    * Also set attribute size, array size and attribute type
  161.    ********************************************************************/
  162.   if (tRecAttr->setup(tAttrInfo, aValue)) {
  163.     //setErrorCodeAbort(4000);
  164.     m_ndb->releaseRecAttr(tRecAttr);
  165.     exit(-1);
  166.     return NULL;
  167.   }
  168.   //theErrorLine++;
  170.   tRecAttr->setNULL();
  171.   
  172.   // We want to keep the list sorted to make data insertion easier later
  173.   if (theFirstRecAttr == NULL) {
  174.     theFirstRecAttr = tRecAttr;
  175.     theCurrentRecAttr = tRecAttr;
  176.     tRecAttr->next(NULL);
  177.   } else {
  178.     Uint32 tAttrId = tAttrInfo->m_attrId;
  179.     if (tAttrId > theCurrentRecAttr->attrId()) { // right order
  180.       theCurrentRecAttr->next(tRecAttr);
  181.       tRecAttr->next(NULL);
  182.       theCurrentRecAttr = tRecAttr;
  183.     } else if (theFirstRecAttr->next() == NULL ||    // only one in list
  184.        theFirstRecAttr->attrId() > tAttrId) {// or first 
  185.       tRecAttr->next(theFirstRecAttr);
  186.       theFirstRecAttr = tRecAttr;
  187.     } else { // at least 2 in list and not first and not last
  188.       NdbRecAttr *p = theFirstRecAttr;
  189.       NdbRecAttr *p_next = p->next();
  190.       while (tAttrId > p_next->attrId()) {
  191. p = p_next;
  192. p_next = p->next();
  193.       }
  194.       if (tAttrId == p_next->attrId()) { // Using same attribute twice
  195. tRecAttr->release(); // do I need to do this?
  196. m_ndb->releaseRecAttr(tRecAttr);
  197. exit(-1);
  198. return NULL;
  199.       }
  200.       // this is it, between p and p_next
  201.       p->next(tRecAttr);
  202.       tRecAttr->next(p_next);
  203.     }
  204.   }
  206.   return tRecAttr;
  207. }
  209. int
  210. NdbEventOperationImpl::execute()
  211. {
  212.   NdbDictionary::Dictionary *myDict = m_ndb->getDictionary();
  213.   if (!myDict) {
  214.     ndbout_c("NdbEventOperation::execute(): getDictionary=NULL");
  215.     return 0;
  216.   }
  218.   if (theFirstRecAttrs[0] == NULL) { // defaults to get all
  219.     
  220.   }
  222.   NdbDictionaryImpl & myDictImpl = NdbDictionaryImpl::getImpl(*myDict);
  225.   int hasSubscriber;
  226.   m_bufferId = 
  227.     m_bufferHandle->prepareAddSubscribeEvent(m_eventImpl->m_eventId,
  228.      hasSubscriber /* return value */);
  230.   m_eventImpl->m_bufferId = m_bufferId;
  232.   int r = -1;
  233.   if (m_bufferId >= 0) {
  234.     // now we check if there's already a subscriber
  236.     if (hasSubscriber == 0) { // only excute if there's no other subscribers 
  237.       r = myDictImpl.executeSubscribeEvent(*m_eventImpl);
  238.     } else {
  239.       r = 0;
  240.     }
  241.     if (r) {
  242.       //Error
  243.       m_bufferHandle->unprepareAddSubscribeEvent(m_bufferId);
  244.       m_state = NdbEventOperation::ERROR;
  245.     } else {
  246.       m_bufferHandle->addSubscribeEvent(m_bufferId, this);
  247.       m_state = NdbEventOperation::EXECUTING;
  248.     }
  249.   } else {
  250.     //Error
  251.     m_state = NdbEventOperation::ERROR;
  252.   }
  253.   return r;
  254. }
  256. int
  257. NdbEventOperationImpl::stop()
  258. {
  259.   if (m_state != NdbEventOperation::EXECUTING)
  260.     return -1;
  262.   //  ndbout_c("NdbEventOperation::stopping()");
  264.   NdbDictionary::Dictionary *myDict = m_ndb->getDictionary();
  265.   if (!myDict) {
  266.     ndbout_c("NdbEventOperation::stop(): getDictionary=NULL");
  267.     return 0;
  268.   }
  270.   NdbDictionaryImpl & myDictImpl = NdbDictionaryImpl::getImpl(*myDict);
  272.   int hasSubscriber;
  273.   int ret = 
  274.     m_bufferHandle->prepareDropSubscribeEvent(m_bufferId,
  275.       hasSubscriber /* return value */);
  277.   if (ret < 0) {
  278.     ndbout_c("prepareDropSubscribeEvent failed");
  279.     return -1;
  280.   }
  281.   //  m_eventImpl->m_bufferId = m_bufferId;
  283.   int r = -1;
  285.   if (hasSubscriber == 0) { // only excute if there's no other subscribers
  286.     r = myDictImpl.stopSubscribeEvent(*m_eventImpl);
  287. #ifdef EVENT_DEBUG
  288.     ndbout_c("NdbEventOperation::stopping() done");
  289. #endif
  290.   } else
  291.     r = 0;
  293.   if (r) {
  294.     //Error
  295.     m_bufferHandle->unprepareDropSubscribeEvent(m_bufferId);
  296.     m_state = NdbEventOperation::ERROR;
  297.   } else {
  298. #ifdef EVENT_DEBUG
  299.     ndbout_c("NdbEventOperation::dropping()");
  300. #endif
  301.     m_bufferHandle->dropSubscribeEvent(m_bufferId);
  302.     m_state = NdbEventOperation::CREATED;
  303.   }
  306.   return r;
  307. }
  309. bool
  310. NdbEventOperationImpl::isConsistent()
  311. {
  312.   return sdata->isGCIConsistent();
  313. }
  315. Uint32
  316. NdbEventOperationImpl::getGCI()
  317. {
  318.   return sdata->gci;
  319. }
  321. Uint32
  322. NdbEventOperationImpl::getLatestGCI()
  323. {
  324.   return NdbGlobalEventBufferHandle::getLatestGCI();
  325. }
  327. int
  328. NdbEventOperationImpl::next(int *pOverrun)
  329. {
  330.   int nr = 10000; // a high value
  331.   int tmpOverrun = 0;
  332.   int *ptmpOverrun;
  333.   if (pOverrun) {
  334.     ptmpOverrun = &tmpOverrun;
  335.   } else
  336.     ptmpOverrun = NULL;
  338.   while (nr > 0) {
  339.     int r=NdbGlobalEventBufferHandle::getDataL(m_bufferId, sdata,
  340.        ptr, pOverrun);
  341.     if (pOverrun) {
  342.       tmpOverrun += *pOverrun;
  343.       *pOverrun = tmpOverrun;
  344.     }
  346.     if (r <= 0) return r; // no data
  348.     if (r < nr) r = nr; else nr--; // we don't want to be stuck here forever
  349.   
  350. #ifdef EVENT_DEBUG
  351.     ndbout_c("!!!!!!!sdata->operation %u", (Uint32)sdata->operation);
  352. #endif
  354.     // now move the data into the RecAttrs
  355.     if ((theFirstRecAttrs[0] == NULL) && 
  356. (theFirstRecAttrs[1] == NULL)) return r;
  357.     // no copying since no RecAttr's
  360.     Uint32 *aAttrPtr = ptr[0].p;
  361.     Uint32 *aAttrEndPtr = aAttrPtr + ptr[0].sz;
  362.     Uint32 *aDataPtr = ptr[1].p;
  364. #ifdef EVENT_DEBUG
  365.     int i;
  366.     printf("after values sz=%un", ptr[1].sz);
  367.     for (i=0; i < ptr[1].sz; i++)
  368.       printf ("H'%.8X ",ptr[1].p[i]);
  369.     printf("n");
  370.     printf("before values sz=%un", ptr[2].sz);
  371.     for (i=0; i < ptr[2].sz; i++)
  372.       printf ("H'%.8X ",ptr[2].p[i]);
  373.     printf("n");
  374. #endif
  376.     NdbRecAttr *tWorkingRecAttr = theFirstRecAttrs[0];
  378.     // copy data into the RecAttr's
  379.     // we assume that the respective attribute lists are sorted
  381.     Uint32 tRecAttrId;
  382.     Uint32 tAttrId;
  383.     Uint32 tDataSz;
  384.     int hasSomeData=0;
  385.     while ((aAttrPtr < aAttrEndPtr) && (tWorkingRecAttr != NULL)) {
  386.       tRecAttrId = tWorkingRecAttr->attrId();
  387.       tAttrId = AttributeHeader(*aAttrPtr).getAttributeId();
  388.       tDataSz = AttributeHeader(*aAttrPtr).getDataSize();
  389.       
  390.       while (tAttrId > tRecAttrId) {
  391. //printf("[%u] %u %u [%u]n", tAttrId, tDataSz, *aDataPtr, tRecAttrId);
  392. tWorkingRecAttr->setNULL();
  393. tWorkingRecAttr = tWorkingRecAttr->next();
  394. if (tWorkingRecAttr == NULL)
  395.   break;
  396. tRecAttrId = tWorkingRecAttr->attrId();
  397.       }
  398.       if (tWorkingRecAttr == NULL)
  399. break;
  400.       
  401.       //printf("[%u] %u %u [%u]n", tAttrId, tDataSz, *aDataPtr, tRecAttrId);
  402.       
  403.       if (tAttrId == tRecAttrId) {
  404. if (!m_eventImpl->m_tableImpl->getColumn(tRecAttrId)->getPrimaryKey())
  405.   hasSomeData++;
  407. //printf("set!n");
  409. tWorkingRecAttr->receive_data(aDataPtr, tDataSz);
  411. // move forward, data has already moved forward
  412. aAttrPtr++;
  413. aDataPtr += tDataSz;
  414. tWorkingRecAttr = tWorkingRecAttr->next();
  415.       } else {
  416. // move only attr forward
  417. aAttrPtr++;
  418. aDataPtr += tDataSz;
  419.       }
  420.     }
  421.     
  422.     while (tWorkingRecAttr != NULL) {
  423.       tRecAttrId = tWorkingRecAttr->attrId();
  424.       //printf("set undefined [%u] %u %u [%u]n", tAttrId, tDataSz, *aDataPtr, tRecAttrId);
  425.       tWorkingRecAttr->setNULL();
  426.       tWorkingRecAttr = tWorkingRecAttr->next();
  427.     }
  428.     
  429.     tWorkingRecAttr = theFirstRecAttrs[1];
  430.     aDataPtr = ptr[2].p;
  431.     Uint32 *aDataEndPtr = aDataPtr + ptr[2].sz;
  432.     while ((aDataPtr < aDataEndPtr) && (tWorkingRecAttr != NULL)) {
  433.       tRecAttrId = tWorkingRecAttr->attrId();
  434.       tAttrId = AttributeHeader(*aDataPtr).getAttributeId();
  435.       tDataSz = AttributeHeader(*aDataPtr).getDataSize();
  436.       aDataPtr++;
  437.       while (tAttrId > tRecAttrId) {
  438. tWorkingRecAttr->setNULL();
  439. tWorkingRecAttr = tWorkingRecAttr->next();
  440. if (tWorkingRecAttr == NULL)
  441.   break;
  442. tRecAttrId = tWorkingRecAttr->attrId();
  443.       }
  444.       if (tWorkingRecAttr == NULL)
  445. break;
  446.       if (tAttrId == tRecAttrId) {
  447. if (!m_eventImpl->m_tableImpl->getColumn(tRecAttrId)->getPrimaryKey())
  448.   hasSomeData++;
  450. tWorkingRecAttr->receive_data(aDataPtr, tDataSz);
  451. aDataPtr += tDataSz;
  452. // move forward, data+attr has already moved forward
  453. tWorkingRecAttr = tWorkingRecAttr->next();
  454.       } else {
  455. // move only data+attr forward
  456. aDataPtr += tDataSz;
  457.       }
  458.     }
  459.     while (tWorkingRecAttr != NULL) {
  460.       tWorkingRecAttr->setNULL();
  461.       tWorkingRecAttr = tWorkingRecAttr->next();
  462.     }
  463.     
  464.     if (hasSomeData)
  465.       return r;
  466.   }
  467.   return 0;
  468. }
  470. NdbDictionary::Event::TableEvent 
  471. NdbEventOperationImpl::getEventType()
  472. {
  473.   switch (sdata->operation) {
  474.   case TriggerEvent::TE_INSERT:
  475.     return NdbDictionary::Event::TE_INSERT;
  476.   case TriggerEvent::TE_DELETE:
  477.     return NdbDictionary::Event::TE_DELETE;
  478.   case TriggerEvent::TE_UPDATE:
  479.     return NdbDictionary::Event::TE_UPDATE;
  480.   default:
  481.     return NdbDictionary::Event::TE_ALL;
  482.   }
  483. }
  487. void
  488. NdbEventOperationImpl::print()
  489. {
  490.   ndbout << "EventId " << m_eventId << "n";
  492.   for (int i = 0; i < 2; i++) {
  493.     NdbRecAttr *p = theFirstRecAttrs[i];
  494.     ndbout << " %u " << i;
  495.     while (p) {
  496.       ndbout << " : " << p->attrId() << " = " << *p;
  497.       p = p->next();
  498.     }
  499.     ndbout << "n";
  500.   }
  501. }
  503. void
  504. NdbEventOperationImpl::printAll()
  505. {
  506.   Uint32 *aAttrPtr = ptr[0].p;
  507.   Uint32 *aAttrEndPtr = aAttrPtr + ptr[0].sz;
  508.   Uint32 *aDataPtr = ptr[1].p;
  510.   //tRecAttr->setup(tAttrInfo, aValue)) {
  512.   Uint32 tAttrId;
  513.   Uint32 tDataSz;
  514.   for (; aAttrPtr < aAttrEndPtr; ) {
  515.     tAttrId = AttributeHeader(*aAttrPtr).getAttributeId();
  516.     tDataSz = AttributeHeader(*aAttrPtr).getDataSize();
  518.     aAttrPtr++;
  519.     aDataPtr += tDataSz;
  520.   }
  521. }
  524. int NdbEventOperationImpl::wait(void *p, int aMillisecondNumber)
  525. {
  526.   return ((NdbGlobalEventBufferHandle*)p)->wait(aMillisecondNumber);
  527. }
  529. /*
  530.  * Global variable ndbGlobalEventBuffer
  531.  * Class NdbGlobalEventBufferHandle
  532.  * Class NdbGlobalEventBuffer
  533.  *
  534.  */
  536. #define ADD_DROP_LOCK_GUARDR(TYPE, FN) 
  537. { 
  538.   ndbGlobalEventBuffer->add_drop_lock(); 
  539.   ndbGlobalEventBuffer->lock(); 
  540.   TYPE r = ndbGlobalEventBuffer->FN; 
  541.   ndbGlobalEventBuffer->unlock(); 
  542.   if (r < 0) { 
  543.     ndbGlobalEventBuffer->add_drop_unlock(); 
  544.   } 
  545.   return r;
  546. }
  547. #define GUARDR(TYPE, FN) 
  548. { 
  549.   ndbGlobalEventBuffer->lock(); 
  550.   TYPE r = ndbGlobalEventBuffer->FN; 
  551.   ndbGlobalEventBuffer->unlock(); 
  552.   return r;
  553. }
  554. #define GUARD(FN) 
  555. { 
  556.   ndbGlobalEventBuffer->lock(); 
  557.   ndbGlobalEventBuffer->FN; 
  558.   ndbGlobalEventBuffer->unlock(); 
  559. }
  560. #define ADD_DROP_UNLOCK_GUARD(FN) 
  561. { 
  562.   GUARD(FN); 
  563.   ndbGlobalEventBuffer->add_drop_unlock(); 
  564. }
  565. #define GUARDBLOCK(BLOCK) 
  566. { 
  567.   ndbGlobalEventBuffer->lock(); 
  568.   BLOCK 
  569.   ndbGlobalEventBuffer->unlock(); 
  570. }
  572. /*
  573.  * Global variable ndbGlobalEventBuffer
  574.  *
  575.  */
  577. extern NdbMutex * ndb_global_event_buffer_mutex;
  578. static NdbGlobalEventBuffer *ndbGlobalEventBuffer=NULL;
  580. /*
  581.  * Class NdbGlobalEventBufferHandle
  582.  * Each Ndb object has a Handle.  This Handle is used to access the
  583.  * global NdbGlobalEventBuffer instance ndbGlobalEventBuffer
  584.  */
  586. NdbGlobalEventBufferHandle *
  587. NdbGlobalEventBuffer_init(int n) 
  588. {
  589.   return new NdbGlobalEventBufferHandle(n);
  590.   // return NdbGlobalEventBufferHandle::init(n);
  591. }
  593. void
  594. NdbGlobalEventBuffer_drop(NdbGlobalEventBufferHandle *h) 
  595. {
  596.   delete h;
  597. }
  599. NdbGlobalEventBufferHandle::NdbGlobalEventBufferHandle
  600. (int MAX_NUMBER_ACTIVE_EVENTS) : m_bufferL(0), m_nids(0)
  601. {
  602.   if ((p_cond = NdbCondition_Create()) ==  NULL) {
  603.     ndbout_c("NdbGlobalEventBufferHandle: NdbCondition_Create() failed");
  604.     exit(-1);
  605.   }
  606.   
  607.   NdbMutex_Lock(ndb_global_event_buffer_mutex);
  608.   if (ndbGlobalEventBuffer == NULL) {
  609.     if (ndbGlobalEventBuffer == NULL) {
  610.       ndbGlobalEventBuffer = new NdbGlobalEventBuffer();
  611.       if (!ndbGlobalEventBuffer) {
  612. NdbMutex_Unlock(ndb_global_event_buffer_mutex);
  613. ndbout_c("NdbGlobalEventBufferHandle:: failed to allocate ndbGlobalEventBuffer");
  614. exit(-1);
  615.       }
  616.     }
  617.   }
  618.   NdbMutex_Unlock(ndb_global_event_buffer_mutex);
  620.   GUARD(real_init(this,MAX_NUMBER_ACTIVE_EVENTS));
  621. }
  623. NdbGlobalEventBufferHandle::~NdbGlobalEventBufferHandle()
  624. {
  625.   NdbCondition_Destroy(p_cond);
  627.   ndbGlobalEventBuffer->lock();
  628.   ndbGlobalEventBuffer->real_remove(this);
  629.   ndbGlobalEventBuffer->unlock();
  631.   NdbMutex_Lock(ndb_global_event_buffer_mutex);
  632.   if (ndbGlobalEventBuffer->m_handlers.size() == 0) {
  633.     delete ndbGlobalEventBuffer;
  634.     ndbGlobalEventBuffer = NULL;
  635.   }
  636.   NdbMutex_Unlock(ndb_global_event_buffer_mutex);
  637. }
  639. void
  640. NdbGlobalEventBufferHandle::addBufferId(int bufferId)
  641. {
  642.   if (m_nids >= NDB_MAX_ACTIVE_EVENTS) {
  643.     ndbout_c("NdbGlobalEventBufferHandle::addBufferId error in paramerer setting");
  644.     exit(-1);
  645.   }
  646.   m_bufferIds[m_nids] = bufferId;
  647.   m_nids++;
  648. }
  650. void
  651. NdbGlobalEventBufferHandle::dropBufferId(int bufferId)
  652. {
  653.   for (int i = 0; i < m_nids; i++)
  654.     if (m_bufferIds[i] == bufferId) {
  655.       m_nids--;
  656.       for (; i < m_nids; i++)
  657. m_bufferIds[i] = m_bufferIds[i+1];
  658.       return;
  659.     }
  660.   ndbout_c("NdbGlobalEventBufferHandle::dropBufferId %d does not exist",
  661.    bufferId);
  662.   exit(-1);
  663. }
  664. /*
  665. NdbGlobalEventBufferHandle *
  666. NdbGlobalEventBufferHandle::init (int MAX_NUMBER_ACTIVE_EVENTS)
  667. {
  668.   return new NdbGlobalEventBufferHandle();
  669. }
  670. void
  671. NdbGlobalEventBufferHandle::drop(NdbGlobalEventBufferHandle *handle)
  672. {
  673.   delete handle;
  674. }
  675. */
  676. int 
  677. NdbGlobalEventBufferHandle::prepareAddSubscribeEvent(Uint32 eventId,
  678.      int& hasSubscriber)
  679. {
  680.   ADD_DROP_LOCK_GUARDR(int,real_prepareAddSubscribeEvent(this, eventId, hasSubscriber));
  681. }
  682. void
  683. NdbGlobalEventBufferHandle::addSubscribeEvent
  684. (int bufferId, NdbEventOperationImpl *ndbEventOperationImpl)
  685. {
  686.   ADD_DROP_UNLOCK_GUARD(real_addSubscribeEvent(bufferId, ndbEventOperationImpl));
  687. }
  688. void
  689. NdbGlobalEventBufferHandle::unprepareAddSubscribeEvent(int bufferId)
  690. {
  691.   ADD_DROP_UNLOCK_GUARD(real_unprepareAddSubscribeEvent(bufferId));
  692. }
  694. int 
  695. NdbGlobalEventBufferHandle::prepareDropSubscribeEvent(int bufferId,
  696.      int& hasSubscriber)
  697. {
  698.   ADD_DROP_LOCK_GUARDR(int,real_prepareDropSubscribeEvent(bufferId, hasSubscriber));
  699. }
  701. void
  702. NdbGlobalEventBufferHandle::unprepareDropSubscribeEvent(int bufferId)
  703. {
  704.   ADD_DROP_UNLOCK_GUARD(real_unprepareDropSubscribeEvent(bufferId));
  705. }
  707. void 
  708. NdbGlobalEventBufferHandle::dropSubscribeEvent(int bufferId)
  709. {
  710.   ADD_DROP_UNLOCK_GUARD(real_dropSubscribeEvent(bufferId));
  711. }
  713. int 
  714. NdbGlobalEventBufferHandle::insertDataL(int bufferId,
  715. const SubTableData * const sdata,
  716. LinearSectionPtr ptr[3])
  717. {
  718.   GUARDR(int,real_insertDataL(bufferId,sdata,ptr));
  719. }
  720.  
  721. void
  722. NdbGlobalEventBufferHandle::latestGCI(int bufferId, Uint32 gci)
  723. {
  724.   GUARD(real_latestGCI(bufferId,gci));
  725. }
  726.  
  727. Uint32
  728. NdbGlobalEventBufferHandle::getLatestGCI()
  729. {
  730.   GUARDR(Uint32, real_getLatestGCI());
  731. }
  732.  
  733. inline void
  734. NdbGlobalEventBufferHandle::group_lock()
  735. {
  736.   ndbGlobalEventBuffer->group_lock();
  737. }
  739. inline void
  740. NdbGlobalEventBufferHandle::group_unlock()
  741. {
  742.   ndbGlobalEventBuffer->group_unlock();
  743. }
  745. int
  746. NdbGlobalEventBufferHandle::wait(int aMillisecondNumber)
  747. {
  748.   GUARDR(int, real_wait(this, aMillisecondNumber));
  749. }
  751. int NdbGlobalEventBufferHandle::getDataL(const int bufferId,
  752.  SubTableData * &sdata,
  753.  LinearSectionPtr ptr[3],
  754.  int *pOverrun)
  755. {
  756.   GUARDR(int,real_getDataL(bufferId,sdata,ptr,pOverrun));
  757. }
  759. /*
  760.  * Class NdbGlobalEventBuffer
  761.  *
  762.  *
  763.  */
  766. void
  767. NdbGlobalEventBuffer::lock()
  768. {
  769.   if (!m_group_lock_flag)
  770.     NdbMutex_Lock(ndb_global_event_buffer_mutex);
  771. }
  772. void
  773. NdbGlobalEventBuffer::unlock()
  774. {
  775.   if (!m_group_lock_flag)
  776.     NdbMutex_Unlock(ndb_global_event_buffer_mutex);
  777. }
  778. void
  779. NdbGlobalEventBuffer::add_drop_lock()
  780. {
  781.   NdbMutex_Lock(p_add_drop_mutex);
  782. }
  783. void
  784. NdbGlobalEventBuffer::add_drop_unlock()
  785. {
  786.   NdbMutex_Unlock(p_add_drop_mutex);
  787. }
  788. inline void
  789. NdbGlobalEventBuffer::group_lock()
  790. {
  791.   lock();
  792.   m_group_lock_flag = 1;
  793. }
  795. inline void
  796. NdbGlobalEventBuffer::group_unlock()
  797. {
  798.   m_group_lock_flag = 0;
  799.   unlock();
  800. }
  802. void
  803. NdbGlobalEventBuffer::lockB(int bufferId)
  804. {
  805.   NdbMutex_Lock(m_buf[ID(bufferId)].p_buf_mutex);
  806. }
  807. void
  808. NdbGlobalEventBuffer::unlockB(int bufferId)
  809. {
  810.   NdbMutex_Lock(m_buf[ID(bufferId)].p_buf_mutex);
  811. }
  813. // Private methods
  815. NdbGlobalEventBuffer::NdbGlobalEventBuffer() : 
  816.   m_handlers(),
  817.   m_group_lock_flag(0),
  818.   m_latestGCI(0),
  819.   m_no(0) // must start at ZERO!
  820. {
  821.   if ((p_add_drop_mutex = NdbMutex_Create()) == NULL) {
  822.     ndbout_c("NdbGlobalEventBuffer: NdbMutex_Create() failed");
  823.     exit(-1);
  824.   }
  825. }
  827. NdbGlobalEventBuffer::~NdbGlobalEventBuffer()
  828. {
  829.   NdbMutex_Destroy(p_add_drop_mutex);
  830.   // NdbMem_Deallocate(m_eventBufferIdToEventId);
  831. }
  832. void
  833. NdbGlobalEventBuffer::real_init (NdbGlobalEventBufferHandle *h, 
  834.  int MAX_NUMBER_ACTIVE_EVENTS)
  835. {
  836.   if (m_handlers.size() == 0) { // First init
  837.     m_max = MAX_NUMBER_ACTIVE_EVENTS;
  838.     m_buf = new BufItem[m_max];
  839.       // (BufItem *)NdbMem_Allocate(m_max*sizeof(BufItem));
  841.     for (int i=0; i<m_max; i++) {
  842.       m_buf[i].gId = 0;
  843.     }
  844.   }
  845.   // TODO make sure we don't hit roof
  846.   //  m_handlers[m_nhandlers] = h;
  847.   m_handlers.push_back(h);
  848.   //  ndbout_c("NdbGlobalEventBuffer::real_init(), m_handles=%u %u", m_nhandlers, h);
  849. }
  850. void
  851. NdbGlobalEventBuffer::real_remove(NdbGlobalEventBufferHandle *h)
  852. {
  853.   //  ndbout_c("NdbGlobalEventBuffer::real_init_remove(), m_handles=%u %u", m_nhandlers, h);
  854.   for (Uint32 i=0 ; i < m_handlers.size(); i++) {
  855.     //    ndbout_c("%u %u %u", i, m_handlers[i], h);
  856.     if (m_handlers[i] == h) {
  857.       m_handlers.erase(i);
  858.       if (m_handlers.size() == 0) {
  859. // ndbout_c("last to go");
  860. delete[] m_buf;
  861. m_buf = NULL;
  862. // NdbMem_Free((char*)m_buf);
  863.       }
  864.       return;
  865.     }
  866.   }
  867.   ndbout_c("NdbGlobalEventBuffer::real_init_remove() non-existing handle");
  868.   exit(-1);
  869. }
  871. int 
  872. NdbGlobalEventBuffer::real_prepareAddSubscribeEvent
  873. (NdbGlobalEventBufferHandle *aHandle, Uint32 eventId, int& hasSubscriber)
  874. {
  875.   int i;
  876.   int bufferId = -1;
  878.   //  add_drop_lock(); // only one thread can do add or drop at a time
  880.   // Find place where eventId already set
  881.   for (i=0; i<m_no; i++) {
  882.     if (m_buf[i].gId == eventId) {
  883.       bufferId = i;
  884.       break;
  885.     }
  886.   }
  887.   if (bufferId < 0) {
  888.     // find space for new bufferId
  889.     for (i=0; i<m_no; i++) {
  890.       if (m_buf[i].gId == 0) {
  891. bufferId = i; // we found an empty spot
  892. break;
  893.       }
  894.     }
  895.     if (bufferId < 0 &&
  896. m_no < m_max) {
  897.       // room for more so get that
  898.       bufferId=m_no;
  899.       m_buf[m_no].gId = 0;
  900.       m_no++;
  901.     } else {
  902.       ndbout_c("prepareAddSubscribeEvent: Can't accept more subscribers");
  903.       //      add_drop_unlock();
  904.       return -1;
  905.     }
  906.   }
  908.   BufItem &b = m_buf[ID(bufferId)];
  910.   if (b.gId == 0) { // first subscriber needs some initialization
  912.     bufferId = NO_ID(0, bufferId);
  914.     b.gId = eventId;
  916.     if ((b.p_buf_mutex = NdbMutex_Create()) == NULL) {
  917.       ndbout_c("NdbGlobalEventBuffer: NdbMutex_Create() failed");
  918.       exit(-1);
  919.     }
  921.     b.subs = 0;
  922.     b.f = 0;
  923.     b.sz = 0;
  924.     b.max_sz = aHandle->m_bufferL;
  925.     b.data = 
  926.       (BufItem::Data *)NdbMem_Allocate(b.max_sz*sizeof(BufItem::Data));
  927.     for (int i = 0; i < b.max_sz; i++) {
  928.       b.data[i].sdata = NULL;
  929.       b.data[i].ptr[0].p = NULL;
  930.       b.data[i].ptr[1].p = NULL;
  931.       b.data[i].ptr[2].p = NULL;
  932.     }
  933.   } else {
  934. #ifdef EVENT_DEBUG
  935.     ndbout_c("NdbGlobalEventBuffer::prepareAddSubscribeEvent: TRYING handle one subscriber per event b.subs = %u", b.subs);
  936. #endif
  938.     int ni = -1;
  939.     for(int i=0; i < b.subs;i++) {
  940.       if (b.ps[i].theHandle == NULL) {
  941. ni = i;
  942. break;
  943.       }
  944.     }
  945.     if (ni < 0) {
  946.       if (b.subs < MAX_SUBSCRIBERS_PER_EVENT) {
  947. ni = b.subs;
  948.       } else {
  949. ndbout_c("prepareAddSubscribeEvent: Can't accept more subscribers");
  950. // add_drop_unlock();
  951. return -1;
  952.       }
  953.     }
  954.     bufferId = NO_ID(ni, bufferId);
  955.   }
  957.   // initialize BufItem::Ps
  958.   {
  959.     int n = NO(bufferId);
  960.     NdbGlobalEventBuffer::BufItem::Ps &e = b.ps[n];
  961.     e.theHandle = aHandle;
  962.     e.b=0;
  963.     e.bufferempty = 1;
  964.     e.overrun=0; // set to -1 to handle first insert
  965.   }
  967.   if (b.subs > 0)
  968.     hasSubscriber = 1;
  969.   else
  970.     hasSubscriber = 0;
  972. #ifdef EVENT_DEBUG
  973.   ndbout_c("prepareAddSubscribeEvent: handed out bufferId %d for eventId %d",
  974.    bufferId, eventId);
  975. #endif
  977.   /* we now have a lock on the prepare so that no one can mess with this
  978.    * unlock comes in unprepareAddSubscribeEvent or addSubscribeEvent
  979.    */
  980.   return bufferId;
  981. }
  983. void
  984. NdbGlobalEventBuffer::real_unprepareAddSubscribeEvent(int bufferId)
  985. {
  986.   BufItem &b = m_buf[ID(bufferId)];
  987.   int n = NO(bufferId);
  989.   b.ps[n].theHandle = NULL;
  991.   // remove subscribers from the end,
  992.   // we have to keep gaps since the position
  993.   // has been handed out in bufferId
  994.   for (int i = b.subs-1; i >= 0; i--)
  995.     if (b.ps[i].theHandle == NULL)
  996.       b.subs--;
  997.     else
  998.       break;
  1000.   if (b.subs == 0) {
  1001. #ifdef EVENT_DEBUG
  1002.     ndbout_c("unprepareAddSubscribeEvent: no more subscribers left on eventId %d", b.gId);
  1003. #endif
  1004.     b.gId = 0;  // We don't have any subscribers, reuse BufItem
  1005.     if (b.data) {
  1006.       NdbMem_Free((void *)b.data);
  1007.       b.data = NULL;
  1008.     }
  1009.     if (b.p_buf_mutex) {
  1010.       NdbMutex_Destroy(b.p_buf_mutex);
  1011.       b.p_buf_mutex = NULL;
  1012.     }
  1013.   }
  1014.   //  add_drop_unlock();
  1015. }
  1017. void
  1018. NdbGlobalEventBuffer::real_addSubscribeEvent(int bufferId, 
  1019.      void *ndbEventOperation)
  1020. {
  1021.   BufItem &b = m_buf[ID(bufferId)];
  1022.   int n = NO(bufferId);
  1024.   b.subs++;
  1025.   b.ps[n].theHandle->addBufferId(bufferId);
  1027.   //  add_drop_unlock();
  1028. #ifdef EVENT_DEBUG
  1029.   ndbout_c("addSubscribeEvent:: added bufferId %d", bufferId);
  1030. #endif
  1031. }
  1033. void
  1034. NdbGlobalEventBuffer::real_unprepareDropSubscribeEvent(int bufferId)
  1035. {
  1036.   //  add_drop_unlock(); // only one thread can do add or drop at a time
  1037. }
  1039. int 
  1040. NdbGlobalEventBuffer::real_prepareDropSubscribeEvent(int bufferId,
  1041.      int& hasSubscriber)
  1042. {
  1043.   //  add_drop_lock(); // only one thread can do add or drop at a time
  1045.   BufItem &b = m_buf[ID(bufferId)];
  1047.   int n = 0;
  1048.   for(int i=0; i < b.subs;i++) {
  1049.     if (b.ps[i].theHandle != NULL)
  1050.       n++;
  1051.   }
  1053.   if (n > 1)
  1054.     hasSubscriber = 1;
  1055.   else if (n == 1)
  1056.     hasSubscriber = 0;
  1057.   else
  1058.     return -1;
  1060.   return 0;
  1061. }
  1063. void
  1064. NdbGlobalEventBuffer::real_dropSubscribeEvent(int bufferId)
  1065. {
  1066.   //  add_drop_lock(); // only one thread can do add-drop at a time
  1068.   BufItem &b = m_buf[ID(bufferId)];
  1069.   int n = NO(bufferId);
  1071.   b.ps[n].overrun=0;
  1072.   b.ps[n].bufferempty=1;
  1073.   b.ps[n].b=0;
  1074.   b.ps[n].theHandle->dropBufferId(bufferId);
  1076.   real_unprepareAddSubscribeEvent(bufferId); // does add_drop_unlock();
  1078. #ifdef EVENT_DEBUG
  1079.   ndbout_c("dropSubscribeEvent:: dropped bufferId %d", bufferId);
  1080. #endif
  1081. }
  1083. void
  1084. NdbGlobalEventBuffer::real_latestGCI(int bufferId, Uint32 gci)
  1085. {
  1086.   if (gci > m_latestGCI)
  1087.     m_latestGCI = gci;
  1088.   else if ((m_latestGCI-gci) > 0xffff) // If NDB stays up :-)
  1089.     m_latestGCI = gci;
  1090. }
  1092. Uint32
  1093. NdbGlobalEventBuffer::real_getLatestGCI()
  1094. {
  1095.   return m_latestGCI;
  1096. }
  1098. int
  1099. NdbGlobalEventBuffer::real_insertDataL(int bufferId, 
  1100.        const SubTableData * const sdata, 
  1101.        LinearSectionPtr ptr[3])
  1102. {
  1103.   BufItem &b = m_buf[ID(bufferId)];
  1104. #ifdef EVENT_DEBUG
  1105.   int n = NO(bufferId);
  1106. #endif
  1107.   {
  1108.     if (b.subs) {
  1109. #ifdef EVENT_DEBUG
  1110.       ndbout_c("data insertion in buffer %d with eventId %d", bufferId, b.gId);
  1111. #endif
  1112.       // move front forward
  1113.       if (copy_data_alloc(sdata, ptr,
  1114.   b.data[b.f].sdata, b.data[b.f].ptr))
  1115. return -1;
  1116.       for (int i=0; i < b.subs; i++) {
  1117. NdbGlobalEventBuffer::BufItem::Ps &e = b.ps[i];
  1118. if (e.theHandle) { // active subscriber
  1119.   if (b.f == e.b) { // next-to-read == written
  1120.     if (e.bufferempty == 0) {
  1121.       e.overrun++; // another item has been overwritten
  1122.       e.b++; // move next-to-read next since old item was overwritten
  1123.       if (e.b == b.max_sz) e.b = 0; // start from beginning
  1124.     }
  1125.   }
  1126.   e.bufferempty = 0;
  1127.   // signal subscriber that there's more to get
  1128.   NdbCondition_Signal(e.theHandle->p_cond);
  1129. }
  1130.       }
  1131.       b.f++; // move next-to-write
  1132.       if (b.f == b.max_sz) b.f = 0; // start from beginning
  1133. #ifdef EVENT_DEBUG
  1134.       ndbout_c("Front= %d Back = %d overun = %d", b.f,
  1135.        b.ps[n].b, b.ps[n].overrun);
  1136. #endif
  1137.     } else {
  1138. #ifdef EVENT_DEBUG
  1139.       ndbout_c("Data arrived before ready eventId", b.gId);
  1140. #endif
  1141.     }
  1142.   }
  1143.   return 0;
  1144. }
  1146. int NdbGlobalEventBuffer::hasData(int bufferId) {
  1147.   BufItem &b = m_buf[ID(bufferId)];
  1148.   int n = NO(bufferId);
  1149.   NdbGlobalEventBuffer::BufItem::Ps &e = b.ps[n];
  1151.   if(e.bufferempty)
  1152.     return 0;
  1154.   if (b.f <= e.b)
  1155.     return b.max_sz-e.b + b.f;
  1156.   else
  1157.     return b.f-e.b;
  1158. }
  1160. int NdbGlobalEventBuffer::real_getDataL(const int bufferId,
  1161. SubTableData * &sdata,
  1162. LinearSectionPtr ptr[3],
  1163. int *pOverrun)
  1164. {
  1165.   BufItem &b = m_buf[ID(bufferId)];
  1166.   int n = NO(bufferId);
  1167.   NdbGlobalEventBuffer::BufItem::Ps &e = b.ps[n];
  1169.   if (pOverrun) {
  1170.     *pOverrun = e.overrun;
  1171.     e.overrun = 0; // if pOverrun is returned to user reset e.overrun
  1172.   }
  1174.   if (e.bufferempty)
  1175.     return 0; // nothing to get
  1177.   if (copy_data_alloc(b.data[e.b].sdata, b.data[e.b].ptr,
  1178.       sdata, ptr))
  1179.     return -1;
  1181.   e.b++; if (e.b == b.max_sz) e.b = 0; // move next-to-read forward
  1183.   if (b.f == e.b) // back has cought up with front
  1184.     e.bufferempty = 1;
  1186. #ifdef EVENT_DEBUG
  1187.   ndbout_c("getting data from buffer %d with eventId %d", bufferId, b.gId);
  1188. #endif
  1190.   return hasData(bufferId)+1;
  1191. }
  1192. int 
  1193. NdbGlobalEventBuffer::copy_data_alloc(const SubTableData * const f_sdata,
  1194.       LinearSectionPtr f_ptr[3],
  1195.       SubTableData * &t_sdata,
  1196.       LinearSectionPtr t_ptr[3])
  1197. {
  1198.   if (t_sdata == NULL) {
  1199.     t_sdata = (SubTableData *)NdbMem_Allocate(sizeof(SubTableData));
  1200.   }
  1201.   memcpy(t_sdata,f_sdata,sizeof(SubTableData));
  1202.   for (int i = 0; i < 3; i++) {
  1203.     LinearSectionPtr & f_p = f_ptr[i];
  1204.     LinearSectionPtr & t_p = t_ptr[i];
  1205.     if (f_p.sz > 0) {
  1206.       if (t_p.p == NULL) {
  1207. t_p.p = (Uint32 *)NdbMem_Allocate(sizeof(Uint32)*f_p.sz);
  1208.       } else if (t_p.sz != f_p.sz) {
  1209. NdbMem_Free(t_p.p);
  1210. t_p.p = (Uint32 *)NdbMem_Allocate(sizeof(Uint32)*f_p.sz);
  1211.       }
  1212.       memcpy(t_p.p, f_p.p, sizeof(Uint32)*f_p.sz);
  1213.     } else if (t_p.p != NULL) {
  1214.       NdbMem_Free(t_p.p);
  1215.       t_p.p = NULL;
  1216.     }
  1217.     t_p.sz = f_p.sz;
  1218.   }
  1219.   return 0;
  1220. }
  1221. int
  1222. NdbGlobalEventBuffer::real_wait(NdbGlobalEventBufferHandle *h,
  1223. int aMillisecondNumber)
  1224. {
  1225.   // check if there are anything in any of the buffers
  1226.   int i;
  1227.   int n = 0;
  1228.   for (i = 0; i < h->m_nids; i++)
  1229.     n += hasData(h->m_bufferIds[i]);
  1230.   if (n) return n;
  1232.   int r = NdbCondition_WaitTimeout(h->p_cond, ndb_global_event_buffer_mutex,
  1233.    aMillisecondNumber);
  1234.   if (r > 0)
  1235.     return -1;
  1237.   n = 0;
  1238.   for (i = 0; i < h->m_nids; i++)
  1239.     n += hasData(h->m_bufferIds[i]);
  1240.   return n;
  1241. }
  1243. template class Vector<NdbGlobalEventBufferHandle*>;