List:Commits« Previous MessageNext Message »
From:Jonas Oreland Date:November 1 2010 10:36am
Subject:bzr commit into mysql-5.1-telco-7.1 branch (jonas:3931)
View as plain text  
#At file:///home/jonas/src/telco-7.1/ based on revid:jonas@perch-20101029122637-4ir05coj8o4n62i3

 3931 Jonas Oreland	2010-11-01 [merge]
      ndb - merge 70 to 71

    modified:
      storage/ndb/include/kernel/signaldata/SumaImpl.hpp
      storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp
      storage/ndb/src/kernel/blocks/suma/Suma.cpp
      storage/ndb/src/kernel/blocks/suma/Suma.hpp
      storage/ndb/src/kernel/vm/DLList.hpp
      storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp
      storage/ndb/test/ndbapi/test_event.cpp
      storage/ndb/test/run-test/daily-basic-tests.txt
=== modified file 'storage/ndb/include/kernel/signaldata/SumaImpl.hpp'
--- a/storage/ndb/include/kernel/signaldata/SumaImpl.hpp	2010-08-26 12:33:33 +0000
+++ b/storage/ndb/include/kernel/signaldata/SumaImpl.hpp	2010-11-01 10:11:47 +0000
@@ -445,6 +445,7 @@ struct SubRemoveRef {
     Busy = 701,
     NoSuchSubscription = 1407,
     Locked = 1411,
+    Defining = 1418,
     AlreadyDropped = 1419
     ,NotStarted = 1428
   };

=== modified file 'storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp'
--- a/storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp	2010-10-21 12:02:45 +0000
+++ b/storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp	2010-11-01 10:14:05 +0000
@@ -545,15 +545,15 @@ void Dbdict::packTableIntoPages(Signal* 
   const Uint32 type= signal->theData[2];
   const Uint32 pageId= signal->theData[3];
 
-  {
-    Uint32 transId = c_retrieveRecord.schemaTransId;
-    GetTabInfoReq req_copy;
-    req_copy.senderRef = c_retrieveRecord.blockRef;
-    req_copy.senderData = c_retrieveRecord.m_senderData;
-    req_copy.schemaTransId = c_retrieveRecord.schemaTransId;
-    req_copy.requestType = c_retrieveRecord.requestType;
-    req_copy.tableId = tableId;
+  Uint32 transId = c_retrieveRecord.schemaTransId;
+  GetTabInfoReq req_copy;
+  req_copy.senderRef = c_retrieveRecord.blockRef;
+  req_copy.senderData = c_retrieveRecord.m_senderData;
+  req_copy.schemaTransId = c_retrieveRecord.schemaTransId;
+  req_copy.requestType = c_retrieveRecord.requestType;
+  req_copy.tableId = tableId;
 
+  {
     SchemaFile::TableEntry *objEntry = 0;
     if(tableId != RNIL)
     {
@@ -610,6 +610,15 @@ void Dbdict::packTableIntoPages(Signal* 
     jam();
     TableRecordPtr tablePtr;
     c_tableRecordPool.getPtr(tablePtr, tableId);
+    if (tablePtr.p->m_obj_ptr_i == RNIL)
+    {
+      jam();
+      sendGET_TABINFOREF(signal, &req_copy,
+                         GetTabInfoRef::TableNotDefined, __LINE__);
+      initRetrieveRecord(0, 0, 0);
+      return;
+    }
+
     packTableIntoPages(w, tablePtr, signal);
     if (unlikely(signal->theData[0] != 0))
     {
@@ -7089,6 +7098,7 @@ void Dbdict::releaseTableObject(Uint32 t
   {
     jam();
     release_object(tablePtr.p->m_obj_ptr_i);
+    tablePtr.p->m_obj_ptr_i = RNIL;
   }
   else
   {

=== modified file 'storage/ndb/src/kernel/blocks/suma/Suma.cpp'
--- a/storage/ndb/src/kernel/blocks/suma/Suma.cpp	2010-10-04 11:13:22 +0000
+++ b/storage/ndb/src/kernel/blocks/suma/Suma.cpp	2010-11-01 10:11:47 +0000
@@ -1329,6 +1329,75 @@ Suma::execSIGNAL_DROPPED_REP(Signal* sig
  * Dump state
  *
  */
+static
+const char*
+cstr(Suma::Subscription::State s)
+{
+  switch(s){
+  case Suma::Subscription::UNDEFINED:
+    return "undefined";
+  case Suma::Subscription::DEFINED:
+    return "defined";
+  case Suma::Subscription::DEFINING:
+    return "defining";
+  }
+  return "<unknown>";
+}
+
+static
+const char*
+cstr(Suma::Subscription::TriggerState s)
+{
+  switch(s){
+  case Suma::Subscription::T_UNDEFINED:
+    return "undefined";
+  case Suma::Subscription::T_CREATING:
+    return "creating";
+  case Suma::Subscription::T_DEFINED:
+    return "defined";
+  case Suma::Subscription::T_DROPPING:
+    return "dropping";
+  case Suma::Subscription::T_ERROR:
+    return "error";
+  }
+  return "<uknown>";
+}
+
+static
+const char*
+cstr(Suma::Subscription::Options s)
+{
+  static char buf[256];
+  buf[0] = 0;
+  strcat(buf, "[");
+  if (s & Suma::Subscription::REPORT_ALL)
+    strcat(buf, " reportall");
+  if (s & Suma::Subscription::REPORT_SUBSCRIBE)
+    strcat(buf, " reportsubscribe");
+  if (s & Suma::Subscription::MARKED_DROPPED)
+    strcat(buf, " dropped");
+  if (s & Suma::Subscription::NO_REPORT_DDL)
+    strcat(buf, " noreportddl");
+  strcat(buf, " ]");
+  return buf;
+}
+
+static
+const char*
+cstr(Suma::Table::State s)
+{
+  switch(s){
+  case Suma::Table::UNDEFINED:
+    return "undefined";
+  case Suma::Table::DEFINING:
+    return "defining";
+  case Suma::Table::DEFINED:
+    return "defined";
+  case Suma::Table::DROPPED:
+    return "dropped";
+  }
+  return "<unknown>";
+}
 
 void
 Suma::execDUMP_STATE_ORD(Signal* signal){
@@ -1557,6 +1626,111 @@ Suma::execDUMP_STATE_ORD(Signal* signal)
     sendSignalWithDelay(reference(), GSN_DUMP_STATE_ORD, signal, 100, 2);
     return;
   }
+
+  if (tCase == 8012)
+  {
+    jam();
+    Uint32 bucket = signal->theData[1];
+    KeyTable<Subscription>::Iterator it;
+    if (signal->getLength() == 1)
+    {
+      jam();
+      bucket = 0;
+      infoEvent("-- Starting dump of subscribers --");
+    }
+
+    c_subscriptions.next(bucket, it);
+    const Uint32 RT_BREAK = 16;
+    for(Uint32 i = 0; i<RT_BREAK || it.bucket == bucket; i++)
+    {
+      jam();
+      if(it.curr.i == RNIL)
+      {
+        jam();
+        infoEvent("-- Ending dump of subscribers --");
+        return;
+      }
+
+      Ptr<Subscription> subPtr = it.curr;
+      Ptr<Table> tabPtr;
+      c_tablePool.getPtr(tabPtr, subPtr.p->m_table_ptrI);
+      infoEvent("Subcription %u id: 0x%.8x key: 0x%.8x state: %s",
+                subPtr.i,
+                subPtr.p->m_subscriptionId,
+                subPtr.p->m_subscriptionKey,
+                cstr(subPtr.p->m_state));
+      infoEvent("  trigger state: %s options: %s",
+                cstr(subPtr.p->m_trigger_state),
+                cstr((Suma::Subscription::Options)subPtr.p->m_options));
+      infoEvent("  tablePtr: %u tableId: %u schemaVersion: 0x%.8x state: %s",
+                tabPtr.i,
+                subPtr.p->m_tableId,
+                tabPtr.p->m_schemaVersion,
+                cstr(tabPtr.p->m_state));
+      {
+        Ptr<Subscriber> ptr;
+        LocalDLList<Subscriber> list(c_subscriberPool,
+                                     subPtr.p->m_subscribers);
+        for (list.first(ptr); !ptr.isNull(); list.next(ptr), i++)
+        {
+          jam();
+          infoEvent("  Subscriber [ %x %u %u ]",
+                    ptr.p->m_senderRef,
+                    ptr.p->m_senderData,
+                    subPtr.i);
+        }
+      }
+
+      {
+        Ptr<SubOpRecord> ptr;
+        LocalDLFifoList<SubOpRecord> list(c_subOpPool,
+                                          subPtr.p->m_create_req);
+
+        for (list.first(ptr); !ptr.isNull(); list.next(ptr), i++)
+        {
+          jam();
+          infoEvent("  create [ %x %u ]",
+                    ptr.p->m_senderRef,
+                    ptr.p->m_senderData);
+        }
+      }
+
+      {
+        Ptr<SubOpRecord> ptr;
+        LocalDLFifoList<SubOpRecord> list(c_subOpPool,
+                                          subPtr.p->m_start_req);
+
+        for (list.first(ptr); !ptr.isNull(); list.next(ptr), i++)
+        {
+          jam();
+          infoEvent("  start [ %x %u ]",
+                    ptr.p->m_senderRef,
+                    ptr.p->m_senderData);
+        }
+      }
+
+      {
+        Ptr<SubOpRecord> ptr;
+        LocalDLFifoList<SubOpRecord> list(c_subOpPool,
+                                          subPtr.p->m_stop_req);
+
+        for (list.first(ptr); !ptr.isNull(); list.next(ptr), i++)
+        {
+          jam();
+          infoEvent("  stop [ %u %x %u ]",
+                    ptr.p->m_opType,
+                    ptr.p->m_senderRef,
+                    ptr.p->m_senderData);
+        }
+      }
+      c_subscriptions.next(it);
+    }
+
+    signal->theData[0] = tCase;
+    signal->theData[1] = it.bucket;
+    sendSignalWithDelay(reference(), GSN_DUMP_STATE_ORD, signal, 100, 2);
+    return;
+  }
 }
 
 void Suma::execDBINFO_SCANREQ(Signal *signal)
@@ -2308,6 +2482,12 @@ Suma::execGET_TABINFOREF(Signal* signal)
     jam();
     ndbrequire(false);
   }
+  if (tabPtr.p->m_state == Table::DROPPED)
+  {
+    jam();
+    do_resend_request = 0;
+  }
+
   if (do_resend_request)
   {
     GetTabInfoReq * req = (GetTabInfoReq *)signal->getDataPtrSend();
@@ -2321,7 +2501,12 @@ Suma::execGET_TABINFOREF(Signal* signal)
                         30, GetTabInfoReq::SignalLength);
     return;
   }
+  get_tabinfo_ref_release(signal, tabPtr);
+}
 
+void
+Suma::get_tabinfo_ref_release(Signal* signal, Ptr<Table> tabPtr)
+{
   LocalDLList<Subscription> subList(c_subscriptionPool,
                                     tabPtr.p->m_subscriptions);
   Ptr<Subscription> subPtr;
@@ -2330,6 +2515,8 @@ Suma::execGET_TABINFOREF(Signal* signal)
   {
     jam();
     Ptr<SubOpRecord> ptr;
+    ndbassert(subPtr.p->m_start_req.isEmpty());
+    ndbassert(subPtr.p->m_stop_req.isEmpty());
     LocalDLFifoList<SubOpRecord> list(c_subOpPool, subPtr.p->m_create_req);
     for (list.first(ptr); !ptr.isNull(); )
     {
@@ -2372,6 +2559,13 @@ Suma::execGET_TABINFO_CONF(Signal* signa
   ndbrequire(tabPtr.p->parseTable(ptr, *this));
   releaseSections(handle);
 
+  if (tabPtr.p->m_state == Table::DROPPED)
+  {
+    jam();
+    get_tabinfo_ref_release(signal, tabPtr);
+    return;
+  }
+
   tabPtr.p->m_state = Table::DEFINED;
 
   LocalDLList<Subscription> subList(c_subscriptionPool,
@@ -3038,19 +3232,6 @@ Suma::execCREATE_TRIG_IMPL_REF(Signal* s
     return;
   }
 
-  for (Uint32 i = 0; i<3; i++)
-  {
-    jam();
-    if (subPtr.p->m_triggers[i] == ILLEGAL_TRIGGER_ID)
-    {
-      jam();
-      /**
-       * Wait for more
-       */
-      return;
-    }
-  }
-
   subPtr.p->m_trigger_state = Subscription::T_ERROR;
   drop_triggers(signal, subPtr);
 }
@@ -4603,63 +4784,93 @@ Suma::execDROP_TAB_CONF(Signal *signal)
   }
 
   DBUG_PRINT("info",("drop table id: %d[i=%u]", tableId, tabPtr.i));
-
+  const Table::State old_state = tabPtr.p->m_state;
   tabPtr.p->m_state = Table::DROPPED;
   c_tables.remove(tabPtr);
 
-  if (tabPtr.p->m_subscriptions.isEmpty())
+  if (senderRef != 0)
   {
     jam();
-    tabPtr.p->release(* this);
-    c_tablePool.release(tabPtr);
-    return;
+
+    // dict coordinator sends info to API
+
+    const Uint64 gci = get_current_gci(signal);
+    SubTableData * data = (SubTableData*)signal->getDataPtrSend();
+    data->gci_hi         = Uint32(gci >> 32);
+    data->gci_lo         = Uint32(gci);
+    data->tableId        = tableId;
+    data->requestInfo    = 0;
+    SubTableData::setOperation(data->requestInfo,
+                               NdbDictionary::Event::_TE_DROP);
+    SubTableData::setReqNodeId(data->requestInfo, refToNode(senderRef));
+
+    Ptr<Subscription> subPtr;
+    LocalDLList<Subscription> subList(c_subscriptionPool,
+                                      tabPtr.p->m_subscriptions);
+
+    for (subList.first(subPtr); !subPtr.isNull(); subList.next(subPtr))
+    {
+      jam();
+      if(subPtr.p->m_subscriptionType != SubCreateReq::TableEvent)
+      {
+        jam();
+        continue;
+        //continue in for-loop if the table is not part of
+        //the subscription. Otherwise, send data to subscriber.
+      }
+
+      if (subPtr.p->m_options & Subscription::NO_REPORT_DDL)
+      {
+        jam();
+        continue;
+      }
+
+      Ptr<Subscriber> ptr;
+      LocalDLList<Subscriber> list(c_subscriberPool, subPtr.p->m_subscribers);
+      for(list.first(ptr); !ptr.isNull(); list.next(ptr))
+      {
+        jam();
+        data->senderData= ptr.p->m_senderData;
+        sendSignal(ptr.p->m_senderRef, GSN_SUB_TABLE_DATA, signal,
+                   SubTableData::SignalLength, JBB);
+      }
+    }
   }
 
-  if (senderRef == 0)
+  if (old_state == Table::DEFINING)
   {
     jam();
     return;
   }
-  // dict coordinator sends info to API
-
-  const Uint64 gci = get_current_gci(signal);
-  SubTableData * data = (SubTableData*)signal->getDataPtrSend();
-  data->gci_hi         = Uint32(gci >> 32);
-  data->gci_lo         = Uint32(gci);
-  data->tableId        = tableId;
-  data->requestInfo    = 0;
-  SubTableData::setOperation(data->requestInfo,NdbDictionary::Event::_TE_DROP);
-  SubTableData::setReqNodeId(data->requestInfo, refToNode(senderRef));
-  
-  Ptr<Subscription> subPtr;
-  LocalDLList<Subscription> subList(c_subscriptionPool,
-                                    tabPtr.p->m_subscriptions);
 
-  for (subList.first(subPtr); !subPtr.isNull(); subList.next(subPtr))
+  if (tabPtr.p->m_subscriptions.isEmpty())
   {
     jam();
-    if(subPtr.p->m_subscriptionType != SubCreateReq::TableEvent)
-    {
-      jam();
-      continue;
-      //continue in for-loop if the table is not part of
-      //the subscription. Otherwise, send data to subscriber.
-    }
-
-    if (subPtr.p->m_options & Subscription::NO_REPORT_DDL)
+    tabPtr.p->release(* this);
+    c_tablePool.release(tabPtr);
+    return;
+  }
+  else
+  {
+    /**
+     * check_release_subscription create a subList...
+     *   weirdness below is to make sure that it's not created twice
+     */
+    Ptr<Subscription> subPtr;
     {
-      jam();
-      continue;
+      LocalDLList<Subscription> subList(c_subscriptionPool,
+                                        tabPtr.p->m_subscriptions);
+      subList.first(subPtr);
     }
-
-    Ptr<Subscriber> ptr;
-    LocalDLList<Subscriber> list(c_subscriberPool, subPtr.p->m_subscribers);
-    for(list.first(ptr); !ptr.isNull(); list.next(ptr))
+    while (!subPtr.isNull())
     {
-      jam();
-      data->senderData= ptr.p->m_senderData;
-      sendSignal(ptr.p->m_senderRef, GSN_SUB_TABLE_DATA, signal,
-                 SubTableData::SignalLength, JBB);
+      Ptr<Subscription> tmp = subPtr;
+      {
+        LocalDLList<Subscription> subList(c_subscriptionPool,
+                                          tabPtr.p->m_subscriptions);
+        subList.next(subPtr);
+      }
+      check_release_subscription(signal, tmp);
     }
   }
 }
@@ -4898,9 +5109,11 @@ Suma::execSUB_REMOVE_REQ(Signal* signal)
   switch(subPtr.p->m_state){
   case Subscription::UNDEFINED:
     jam();
+    ndbrequire(false);
   case Subscription::DEFINING:
     jam();
-    ndbrequire(false);
+    sendSubRemoveRef(signal, req, SubRemoveRef::Defining);
+    return;
   case Subscription::DEFINED:
     if (subPtr.p->m_options & Subscription::MARKED_DROPPED)
     {

=== modified file 'storage/ndb/src/kernel/blocks/suma/Suma.hpp'
--- a/storage/ndb/src/kernel/blocks/suma/Suma.hpp	2010-09-29 07:39:42 +0000
+++ b/storage/ndb/src/kernel/blocks/suma/Suma.hpp	2010-11-01 10:11:47 +0000
@@ -412,6 +412,7 @@ public:
   void check_remove_queue(Signal*, Ptr<Subscription>,
                           Ptr<SubOpRecord>,bool,bool);
   void check_release_subscription(Signal* signal, Ptr<Subscription>);
+  void get_tabinfo_ref_release(Signal*, Ptr<Table>);
 
   /**
    * Public interface

=== modified file 'storage/ndb/src/kernel/vm/DLList.hpp'
--- a/storage/ndb/src/kernel/vm/DLList.hpp	2009-10-08 09:55:36 +0000
+++ b/storage/ndb/src/kernel/vm/DLList.hpp	2010-11-01 10:23:18 +0000
@@ -35,7 +35,16 @@ public:
   struct HeadPOD {
     Uint32 firstItem;
     inline bool isEmpty() const { return firstItem == RNIL; }
-    inline void init () { firstItem = RNIL; }
+    inline void init () { 
+      firstItem = RNIL; 
+#ifdef VM_TRACE
+      in_use = false;
+#endif
+    }
+
+#ifdef VM_TRACE
+    bool in_use;
+#endif
   };
 
   struct Head : public HeadPOD 
@@ -169,9 +178,16 @@ public:
     : DLListImpl<P,T,U>(thePool), src(_src)
   {
     this->head = src;
+#ifdef VM_TRACE
+    assert(src.in_use == false);
+    src.in_use = true;
+#endif
   }
   
   ~LocalDLListImpl(){
+#ifdef VM_TRACE
+    assert(src.in_use == true);
+#endif
     src = this->head;
   }
 private:

=== modified file 'storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp'
--- a/storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp	2010-10-06 12:35:34 +0000
+++ b/storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp	2010-11-01 10:11:47 +0000
@@ -1699,7 +1699,7 @@ NdbEventImpl::setTable(NdbTableImpl *tab
 {
   DBUG_ENTER("NdbEventImpl::setTable");
   DBUG_PRINT("info", ("this: %p  tableImpl: %p", this, tableImpl));
-  DBUG_ASSERT(tableImpl->m_status != NdbDictionary::Object::Invalid);
+
   if (!m_tableImpl) 
     m_tableImpl = new NdbTableImpl();
   // Copy table, since event might be accessed from different threads

=== modified file 'storage/ndb/test/ndbapi/test_event.cpp'
--- a/storage/ndb/test/ndbapi/test_event.cpp	2010-09-06 08:20:33 +0000
+++ b/storage/ndb/test/ndbapi/test_event.cpp	2010-11-01 10:11:47 +0000
@@ -3317,6 +3317,66 @@ runBug56579(NDBT_Context* ctx, NDBT_Step
   return result;
 }
 
+int
+runBug57886_create_drop(NDBT_Context* ctx, NDBT_Step* step)
+{
+  int loops = ctx->getNumLoops();
+  Ndb* pNdb = GETNDB(step);
+
+  NdbDictionary::Dictionary *pDict = pNdb->getDictionary();
+  NdbDictionary::Table tab = * ctx->getTab();
+
+  sleep(5);
+
+  while (loops --)
+  {
+    if (pDict->dropTable(tab.getName()) != 0)
+    {
+      return NDBT_FAILED;
+    }
+
+    if (pDict->createTable(tab) != 0)
+    {
+      return NDBT_FAILED;
+    }
+
+    sleep(1);
+  }
+
+  ctx->stopTest();
+  return NDBT_OK;
+}
+
+int
+runBug57886_subscribe_unsunscribe(NDBT_Context* ctx, NDBT_Step* step)
+{
+  Ndb* pNdb;
+  Ndb_cluster_connection *pCC;
+  if (cc(&pCC, &pNdb))
+  {
+    // too few api slots...
+    return NDBT_OK;
+  }
+
+  NdbDictionary::Table tab = * ctx->getTab();
+
+  while (!ctx->isTestStopped())
+  {
+    createEvent(pNdb, tab, false, false);
+
+    NdbEventOperation* op = createEventOperation(pNdb, tab, 0);
+    if (op)
+    {
+      pNdb->dropEventOperation(op);
+    }
+    dropEvent(pNdb, tab);
+  }
+
+  delete pNdb;
+  delete pCC;
+  return NDBT_OK;
+}
+
 NDBT_TESTSUITE(test_event);
 TESTCASE("BasicEventOperation", 
 	 "Verify that we can listen to Events"
@@ -3545,6 +3605,11 @@ TESTCASE("Bug56579", "")
   STEP(runBug56579);
   FINALIZER(runDropEvent);
 }
+TESTCASE("Bug57886", "")
+{
+  STEP(runBug57886_create_drop);
+  STEPS(runBug57886_subscribe_unsunscribe, 5);
+}
 NDBT_TESTSUITE_END(test_event);
 
 int main(int argc, const char** argv){

=== modified file 'storage/ndb/test/run-test/daily-basic-tests.txt'
--- a/storage/ndb/test/run-test/daily-basic-tests.txt	2010-10-28 12:59:31 +0000
+++ b/storage/ndb/test/run-test/daily-basic-tests.txt	2010-11-01 10:11:47 +0000
@@ -1480,6 +1480,10 @@ max-time: 1200
 cmd: test_event
 args: -n Bug56579 T1
 
+max-time: 1200
+cmd: test_event
+args: -n Bug57886 T1
+
 max-time: 3600
 cmd: testNodeRestart
 args: -n Bug44952 T1

No bundle (reason: revision is a merge).
Thread
bzr commit into mysql-5.1-telco-7.1 branch (jonas:3931) Jonas Oreland1 Nov