Bug#32776593 NOT ALL DATA NODE TO DATA NODE CONNECTIONS SEND HEART BEAT MESSAGING [1/3]

zmur · zmur · commit 1f707aa14804 · 2021-06-15T14:29:28.000+02:00
Add method to send a signal on all transporters links to a node.

Change-Id: I132225deae54c1dba2d0b7a02c17a4d7f6298c00
diff --git a/storage/ndb/include/transporter/TransporterRegistry.hpp b/storage/ndb/include/transporter/TransporterRegistry.hpp
@@ -467,9 +467,13 @@ class TransporterRegistry
                          Uint8 prio,
                          const Uint32 *signalData,
                          NodeId nodeId,
-                         TrpId &trp_id,
+                         Transporter* t,
                          AnySectionArg section);
 
+  Transporter* prepareSend_getTransporter(const SignalHeader *signalHeader,
+                                          NodeId nodeId,
+                                          TrpId &trp_id,
+                                          SendStatus& status);
 
 public:
   SendStatus prepareSend(TransporterSendBufferHandle *sendHandle,
@@ -495,7 +499,15 @@ class TransporterRegistry
                          const Uint32 *signalData,
                          NodeId nodeId,
                          const GenericSectionPtr ptr[3]);
-  
+
+  SendStatus prepareSendOverAllLinks(
+                         TransporterSendBufferHandle *sendHandle,
+                         const SignalHeader *signalHeader,
+                         Uint8 prio,
+                         const Uint32 *signalData,
+                         NodeId nodeId,
+                         TrpBitmask &trp_mask);
+
   /* Send on a specific transporter */
   bool performSend(TrpId id, bool need_wakeup = true);
   /* performSendNode is only used from NDB API */
diff --git a/storage/ndb/src/common/transporter/TransporterRegistry.cpp b/storage/ndb/src/common/transporter/TransporterRegistry.cpp
@@ -1058,31 +1058,11 @@ TransporterRegistry::prepareSendTemplate(
                                  Uint8 prio,
                                  const Uint32 * signalData,
                                  NodeId nodeId,
-                                 TrpId &trp_id,
+                                 Transporter *t,
                                  AnySectionArg section)
 {
-  Transporter *node_trp = theNodeIdTransporters[nodeId];
-  if (unlikely(node_trp == NULL))
-  {
-    DEBUG("Discarding message to unknown node: " << nodeId);
-    return SEND_UNKNOWN_NODE;
-  }
-  assert(!node_trp->isPartOfMultiTransporter());
-  Transporter *t;
-  t = node_trp->get_send_transporter(signalHeader->theReceiversBlockNumber,
-                                     signalHeader->theSendersBlockRef);
-  assert(!t->isMultiTransporter());
-  trp_id = t->getTransporterIndex();
-  if (unlikely(trp_id == 0))
-  {
-    /**
-     * Can happen in disconnect situations, node is disconnected, so send
-     * to it is successful since the node won't be there to receive the
-     * message.
-     */
-    DEBUG("Discarding message due to trp_id = 0");
-    return SEND_OK;
-  }
+  assert(t != nullptr);
+
   if(
     likely((ioStates[nodeId] != HaltOutput) && (ioStates[nodeId] != HaltIO)) || 
            (signalHeader->theReceiversBlockNumber == QMGR) ||
@@ -1093,6 +1073,7 @@ TransporterRegistry::prepareSendTemplate(
       const Uint32 lenBytes = t->m_packer.getMessageLength(signalHeader, section.m_ptr);
       if (likely(lenBytes <= MAX_SEND_MESSAGE_BYTESIZE))
       {
+        TrpId trp_id = t->getTransporterIndex();
         SendStatus error = SEND_OK;
 	Uint32 *insertPtr = getWritePtr(sendHandle,
                                         t,
@@ -1184,6 +1165,40 @@ TransporterRegistry::prepareSendTemplate(
 }
 
 
+Transporter*
+TransporterRegistry::prepareSend_getTransporter(
+                                const SignalHeader *signalHeader,
+                                NodeId nodeId,
+                                TrpId &trp_id,
+                                SendStatus& status)
+{
+  Transporter *node_trp = theNodeIdTransporters[nodeId];
+  if (unlikely(node_trp == NULL))
+  {
+    DEBUG("Discarding message to unknown node: " << nodeId);
+    status = SEND_UNKNOWN_NODE;
+    return nullptr;
+  }
+  assert(!node_trp->isPartOfMultiTransporter());
+  Transporter *t;
+  t = node_trp->get_send_transporter(signalHeader->theReceiversBlockNumber,
+                                     signalHeader->theSendersBlockRef);
+  assert(!t->isMultiTransporter());
+  trp_id = t->getTransporterIndex();
+  if (unlikely(trp_id == 0))
+  {
+    /**
+     * Can happen in disconnect situations, node is disconnected, so send
+     * to it is successful since the node won't be there to receive the
+     * message.
+     */
+    DEBUG("Discarding message due to trp_id = 0");
+    status = SEND_OK;
+    return nullptr;
+  }
+  return t;
+}
+
 SendStatus
 TransporterRegistry::prepareSend(TransporterSendBufferHandle *sendHandle,
                                  const SignalHeader *signalHeader,
@@ -1193,13 +1208,19 @@ TransporterRegistry::prepareSend(TransporterSendBufferHandle *sendHandle,
                                  TrpId &trp_id,
                                  const LinearSectionPtr ptr[3])
 {
+  SendStatus status;
+  Transporter *t = prepareSend_getTransporter(signalHeader, nodeId, trp_id,
+                                              status);
+  if (unlikely(t == nullptr))
+    return status;
+
   const Packer::LinearSectionArg section(ptr);
   return prepareSendTemplate(sendHandle,
                              signalHeader,
                              prio,
                              signalData,
                              nodeId,
-                             trp_id,
+                             t,
                              section);
 }
 
@@ -1214,13 +1235,19 @@ TransporterRegistry::prepareSend(TransporterSendBufferHandle *sendHandle,
                                  class SectionSegmentPool &thePool,
                                  const SegmentedSectionPtr ptr[3])
 {
+  SendStatus status;
+  Transporter *t = prepareSend_getTransporter(signalHeader, nodeId, trp_id,
+                                              status);
+  if (unlikely(t == nullptr))
+    return status;
+
   const Packer::SegmentedSectionArg section(thePool,ptr);
   return prepareSendTemplate(sendHandle,
                              signalHeader,
                              prio,
                              signalData,
                              nodeId,
-                             trp_id,
+                             t,
                              section);
 }
 
@@ -1234,16 +1261,114 @@ TransporterRegistry::prepareSend(TransporterSendBufferHandle *sendHandle,
                                  const GenericSectionPtr ptr[3])
 {
   TrpId trp_id = 0;
+  SendStatus status;
+  Transporter *t = prepareSend_getTransporter(signalHeader, nodeId, trp_id,
+                                              status);
+  if (unlikely(t == nullptr))
+    return status;
+
   const Packer::GenericSectionArg section(ptr);
   return prepareSendTemplate(sendHandle,
                              signalHeader,
                              prio,
                              signalData,
                              nodeId,
-                             trp_id,
+                             t,
                              section);
 }
 
+SendStatus
+TransporterRegistry::prepareSendOverAllLinks(
+                                 TransporterSendBufferHandle *sendHandle,
+                                 const SignalHeader *signalHeader,
+                                 Uint8 prio,
+                                 const Uint32 *signalData,
+                                 NodeId nodeId,
+                                 TrpBitmask &trp_ids)
+{
+  // node_trp handling copied from first part of prepareSend_getTransporter
+  Transporter *node_trp = theNodeIdTransporters[nodeId];
+  if (unlikely(node_trp == NULL))
+  {
+    DEBUG("Discarding message to unknown node: " << nodeId);
+    return SEND_UNKNOWN_NODE;
+  }
+  assert(!node_trp->isPartOfMultiTransporter());
+
+  LinearSectionPtr ptr[3];
+  require(signalHeader->m_noOfSections == 0);
+  const Packer::LinearSectionArg section(ptr);
+
+  if (!node_trp->isMultiTransporter())
+  {
+    Transporter *t = node_trp;
+    // t handling copied from second part of prepareSend_getTransporter
+    TrpId trp_id = t->getTransporterIndex();
+    if (unlikely(trp_id == 0))
+    {
+      /**
+       * Can happen in disconnect situations, node is disconnected, so send
+       * to it is successful since the node won't be there to receive the
+       * message.
+       */
+      DEBUG("Discarding message due to trp_id = 0");
+      return SEND_OK;
+    }
+
+    SendStatus status = prepareSendTemplate(sendHandle,
+                                            signalHeader,
+                                            prio,
+                                            signalData,
+                                            nodeId,
+                                            t,
+                                            section);
+
+    if (likely(status == SEND_OK))
+    {
+      require(trp_id < MAX_NTRANSPORTERS);
+      trp_ids.set(trp_id);
+    }
+    return status;
+  }
+  else
+  {
+    Multi_Transporter *multi_trp = get_node_multi_transporter(nodeId);
+    require(multi_trp == node_trp);
+
+    SendStatus return_status = SEND_OK;
+    Uint32 num_trps = multi_trp->get_num_active_transporters();
+    for (Uint32 i = 0; i < num_trps; i++)
+    {
+      Transporter *t = multi_trp->get_active_transporter(i);
+      require(t != nullptr);
+      const TrpId trp_id = t->getTransporterIndex();
+      if (unlikely(trp_id == 0))
+        continue;
+      SendStatus status = prepareSendTemplate(sendHandle,
+                                              signalHeader,
+                                              prio,
+                                              signalData,
+                                              nodeId,
+                                              t,
+                                              section);
+      if (likely(status == SEND_OK))
+      {
+        require(trp_id < MAX_NTRANSPORTERS);
+        trp_ids.set(trp_id);
+      }
+      else if (status != SEND_BLOCKED && status != SEND_DISCONNECTED)
+      {
+        /*
+         * Treat SEND_BLOCKED and SEND_DISCONNECTED as SEND_OK.
+         * Else take the last bad status returned.
+         */
+        return_status = status;
+      }
+    }
+    return return_status;
+  }
+}
+
 bool
 TransporterRegistry::setup_wakeup_socket(TransporterReceiveHandle& recvdata)
 {
diff --git a/storage/ndb/src/kernel/vm/SimulatedBlock.cpp b/storage/ndb/src/kernel/vm/SimulatedBlock.cpp
@@ -1963,6 +1963,95 @@ SimulatedBlock::sendSignalWithDelay(BlockReference ref,
   sections->m_cnt = 0;
 }
 
+/*
+ * Copy implementation for sendSignalOverAllLinks from sendSignal, excluding part for
+ * local send.
+ */
+void
+SimulatedBlock::sendSignalOverAllLinks(BlockReference ref,
+                                 GlobalSignalNumber gsn,
+                                 Signal25* signal,
+                                 Uint32 length,
+                                 JobBufferLevel jobBuffer) const
+{
+
+  BlockReference sendBRef = reference();
+
+  Uint32 recBlock = refToBlock(ref);
+  Uint32 recNode   = refToNode(ref);
+  Uint32 ourProcessor         = globalData.ownId;
+
+  ndbrequire(signal->header.m_noOfSections == 0);
+  check_sections(signal, signal->header.m_noOfSections, 0);
+
+  signal->header.theLength = length;
+  signal->header.theVerId_signalNumber = gsn;
+  signal->header.theReceiversBlockNumber = recBlock;
+  signal->header.m_noOfSections = 0;
+
+  Uint32 tSignalId = signal->header.theSignalId;
+
+  if (unlikely((length == 0) || length > 25 || (recBlock == 0)))
+  {
+    signal_error(gsn, length, recBlock, __FILE__, __LINE__);
+    return;
+  }//if
+#ifdef VM_TRACE
+  if (globalData.testOn){
+    Uint16 proc =
+      (recNode == 0 ? globalData.ownId : recNode);
+    signal->header.theSendersBlockRef = sendBRef;
+    globalSignalLoggers.sendSignal(signal->header,
+                                   jobBuffer,
+                                   &signal->theData[0],
+                                   proc);
+  }
+#endif
+
+  // Local send part not copied from sendSignal.
+  ndbrequire(recNode != ourProcessor);
+  ndbrequire(recNode != 0);
+
+  // send distributed Signal
+  SignalHeader sh;
+
+  Uint32 tTrace = signal->getTrace();
+
+  sh.theVerId_signalNumber   = gsn;
+  sh.theReceiversBlockNumber = recBlock;
+  sh.theSendersBlockRef      = refToBlock(sendBRef);
+  sh.theLength               = length;
+  sh.theTrace                = tTrace;
+  sh.theSignalId             = tSignalId;
+  sh.m_noOfSections          = 0;
+  sh.m_fragmentInfo          = 0;
+
+#ifdef TRACE_DISTRIBUTED
+  g_eventLogger->info("send: %s(%d) to (%s, %d)", getSignalName(gsn), gsn,
+                      getBlockName(recBlock), recNode);
+#endif
+
+  SendStatus ss;
+#ifdef NDBD_MULTITHREADED
+  ss = mt_send_remote_over_all_links(m_threadId, &sh, jobBuffer,
+                                     &signal->theData[0], recNode);
+#else
+  TrpBitmask trp_ids;
+  ss = globalTransporterRegistry.
+         prepareSendOverAllLinks(getNonMTTransporterSendHandle(),
+                                 &sh, jobBuffer,
+                                 &signal->theData[0], recNode, trp_ids);
+#endif
+
+  if (unlikely(! (ss == SEND_OK ||
+                  ss == SEND_BLOCKED ||
+                  ss == SEND_DISCONNECTED)))
+  {
+    handle_send_failed(ss, signal, recNode, (LinearSectionPtr*)NULL);
+  }
+  return;
+}
+
 void
 SimulatedBlock::release(SegmentedSectionPtr & ptr)
 {
diff --git a/storage/ndb/src/kernel/vm/SimulatedBlock.hpp b/storage/ndb/src/kernel/vm/SimulatedBlock.hpp
@@ -842,6 +842,12 @@ class alignas(NDB_CL) SimulatedBlock :
 			   Uint32 length,
 			   SectionHandle* sections) const;
 
+  void sendSignalOverAllLinks(BlockReference ref,
+                        GlobalSignalNumber gsn,
+                        Signal25* signal,
+                        Uint32 length,
+                        JobBufferLevel jbuf ) const ;
+
   /**
    * EXECUTE_DIRECT comes in five variants.
    *
diff --git a/storage/ndb/src/kernel/vm/mt.cpp b/storage/ndb/src/kernel/vm/mt.cpp
diff --git a/storage/ndb/src/kernel/vm/mt.hpp b/storage/ndb/src/kernel/vm/mt.hpp
