Bug#25443080 MAIN THREAD CPU USAGE ON MASTER DATANODE INCREASES TO 100% WITH LATENCY BURST

LCP is controlled by DIH Master on one node, running on the 'main' thread.

CPU usage was seen to be high there towards the end of an LCP, leading to
increased transaction latency.

Problem appears to be inefficiencies in the algorithm used by DIH Master to
find fragment replicas to schedule for LCP, towards the end of the LCP it
can spend a lot of time scanning over the set of all replicas.

Fixes :
 - Make sure that per-node queues are made use of
   - If a fragment replica is queued, do not search for more
   - When searching for more, search for all nodes in 1 pass
 - When a node has started on all of its fragment replicas, do not
   include it in future searches
 - When all nodes have started on all of their fragment replicas,
   no need to search for more.

Testcase :
 - 16 nodes, 4 LDMs/node : 64 fragments : 128 fragment replicas / table
 - 88 tables
 - No data
 - Run LCP using ALL DUMP 7099
 - Observe CPU usage and efficiency of replica search algorithms

Author: frazerclement

storage/ndb/src/kernel/blocks/dbdih/Dbdih.hpp

@@ -1,5 +1,5 @@
 /*
-   Copyright (c) 2003, 2016, Oracle and/or its affiliates. All rights reserved.
+   Copyright (c) 2003, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -2168,6 +2168,7 @@ class Dbdih: public SimulatedBlock {
     SignalCounter m_LAST_LCP_FRAG_ORD;
     NdbNodeBitmask m_participatingLQH;
     NdbNodeBitmask m_participatingDIH;
+    NdbNodeBitmask m_allReplicasQueuedLQH;
 
     Uint32 m_masterLcpDihRef;
     bool   m_MASTER_LCPREQ_Received;

storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp

@@ -1,5 +1,5 @@
 /*
-   Copyright (c) 2003, 2016, Oracle and/or its affiliates. All rights reserved.
+   Copyright (c) 2003, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -18266,6 +18266,13 @@ Dbdih::master_lcp_fragmentMutex_locked(Signal* signal,
   startLcpRoundLoopLab(signal, 0, 0);
 }
 
+
+//#define DIH_DEBUG_REPLICA_SEARCH
+#ifdef DIH_DEBUG_REPLICA_SEARCH
+static Uint32 totalScheduled;
+static Uint32 totalExamined;
+#endif
+
 void Dbdih::startLcpRoundLoopLab(Signal* signal, 
 				 Uint32 startTableId, Uint32 startFragId) 
 {
@@ -18279,19 +18286,53 @@ void Dbdih::startLcpRoundLoopLab(Signal* signal,
   }//if
   c_lcpState.currentFragment.tableId = startTableId;
   c_lcpState.currentFragment.fragmentId = startFragId;
+  c_lcpState.m_allReplicasQueuedLQH.clear();
+
+#ifdef DIH_DEBUG_REPLICA_SEARCH
+  totalScheduled = totalExamined = 0;
+#endif
+
   startNextChkpt(signal);
 }//Dbdih::startLcpRoundLoopLab()
 
 void Dbdih::startNextChkpt(Signal* signal)
 {
+  jam();
+  const bool allReplicaCheckpointsQueued = 
+    c_lcpState.m_allReplicasQueuedLQH.
+    contains(c_lcpState.m_participatingLQH);
+  
+  if (allReplicaCheckpointsQueued)
+  {
+    jam();
+
+    /**
+     * No need to find new checkpoints to start,
+     * just waiting for completion
+     */
+
+    sendLastLCP_FRAG_ORD(signal);
+    return;
+  }
+
   Uint32 lcpId = SYSFILE->latestLCP_ID;
 
-  NdbNodeBitmask busyNodes; 
-  busyNodes.clear();
+  /* Initialise handledNodes with those already fully queued */
+  NdbNodeBitmask handledNodes = c_lcpState.m_allReplicasQueuedLQH; 
+  
+  /* Remove any that have failed in the interim */
+  handledNodes.bitAND(c_lcpState.m_participatingLQH);
+  
   const Uint32 lcpNodes = c_lcpState.m_participatingLQH.count();
 
   bool save = true;
   LcpState::CurrentFragment curr = c_lcpState.currentFragment;
+
+#ifdef DIH_DEBUG_REPLICA_SEARCH
+  Uint32 examined = 0;
+  Uint32 started = 0;
+  Uint32 queued = 0;
+#endif
 
   while (curr.tableId < ctabFileSize) {
     TabRecordPtr tabPtr;
@@ -18314,6 +18355,10 @@ void Dbdih::startNextChkpt(Signal* signal)
 
       jam();
       c_replicaRecordPool.getPtr(replicaPtr);
+
+#ifdef DIH_DEBUG_REPLICA_SEARCH
+      examined++;
+#endif
 
       NodeRecordPtr nodePtr;
       nodePtr.i = replicaPtr.p->procNode;
@@ -18350,6 +18395,10 @@ void Dbdih::startNextChkpt(Signal* signal)
 	    nodePtr.p->noOfStartedChkpt = i + 1;
 
 	    sendLCP_FRAG_ORD(signal, nodePtr.p->startedChkpt[i]);
+
+#ifdef DIH_DEBUG_REPLICA_SEARCH
+            started++;
+#endif
 	  } 
           else if (nodePtr.p->noOfQueuedChkpt <
                    MAX_QUEUED_FRAG_CHECKPOINTS_PER_NODE)
@@ -18369,30 +18418,42 @@ void Dbdih::startNextChkpt(Signal* signal)
 	    nodePtr.p->queuedChkpt[i].fragId = curr.fragmentId;
 	    nodePtr.p->queuedChkpt[i].replicaPtr = replicaPtr.i;
 	    nodePtr.p->noOfQueuedChkpt = i + 1;
+#ifdef DIH_DEBUG_REPLICA_SEARCH
+            queued++;
+#endif
 	  }
 	  else 
 	  {
 	    jam();
 
 	    if(save)
 	    {
-	      /**
-	       * Stop increasing value on first that was "full"
-	       */
-	      c_lcpState.currentFragment = curr;
-	      save = false;
-	    }
+              /**
+               * Stop increasing value on first replica that 
+               * we could not enqueue, so we don't miss it 
+               * next time
+               */
+              c_lcpState.currentFragment = curr;
+              save = false;
+            }
 
-	    busyNodes.set(nodePtr.i);
-	    if(busyNodes.count() == lcpNodes)
+	    handledNodes.set(nodePtr.i);
+	    if (handledNodes.count() == lcpNodes)
 	    {
-	      /**
-	       * There were no possibility to start the local checkpoint 
-	       * and it was not possible to queue it up. In this case we 
-	       * stop the start of local checkpoints until the nodes with a 
-	       * backlog have performed more checkpoints. We will return and 
-	       * will not continue the process of starting any more checkpoints.
-	       */
+              /**
+               * All participating nodes have either
+               * - Full queues
+               * - All available replica checkpoints queued
+               *   (m_allReplicasQueuedLQH)
+               *
+               * Therefore, exit the search here.
+               */
+#ifdef DIH_DEBUG_REPLICA_SEARCH
+              ndbout_c("Search : All nodes busy.  Examined %u Started %u Queued %u",
+                       examined, started, queued);
+              totalExamined+= examined;
+              totalScheduled += (started + queued);
+#endif
 	      return;
 	    }//if
 	  }//if
@@ -18406,6 +18467,31 @@ void Dbdih::startNextChkpt(Signal* signal)
       curr.tableId++;
     }//if
   }//while
+
+#ifdef DIH_DEBUG_REPLICA_SEARCH
+  ndbout_c("Search : At least one node not busy.  Examined %u Started %u Queued %u",
+           examined, started, queued);
+  totalExamined+= examined;
+  totalScheduled += (started + queued);
+#endif
+
+  /**
+   * Have examined all replicas and attempted to 
+   * enqueue as many replica LCPs as possible,
+   * without filling all queues.
+   * This means that some node(s) have no more
+   * replica LCPs to be enqueued.
+   * These are the node(s) which are *not* in
+   * the handled bitmap on this round.
+   * We keep track of these to allow the search
+   * to exit early on future invocations.
+   */
+  
+  /* Invert handled nodes to reveal newly finished nodes */
+  handledNodes.bitXOR(c_lcpState.m_participatingLQH);
+  
+  /* Add newly finished nodes to the global state */
+  c_lcpState.m_allReplicasQueuedLQH.bitOR(handledNodes);
 
   sendLastLCP_FRAG_ORD(signal);
 }//Dbdih::startNextChkpt()
@@ -18874,6 +18960,21 @@ Dbdih::checkLcpAllTablesDoneInLqh(Uint32 line){
     ndbout_c("CLEARING 7194");
     CLEAR_ERROR_INSERT_VALUE;
   }
+
+#ifdef DIH_DEBUG_REPLICA_SEARCH
+  if (totalScheduled == 0)
+  {
+    totalScheduled = 1;
+  }
+  ndbout_c("LCP complete.  Examined %u replicas, scheduled %u.  Ratio : %u.%u",
+           totalExamined,
+           totalScheduled,
+           totalExamined/totalScheduled,
+           (10 * (totalExamined - 
+                  ((totalExamined/totalScheduled) *
+                   totalScheduled)))/
+           totalScheduled);
+#endif
 
   return true;
 }
@@ -19059,13 +19160,14 @@ void Dbdih::checkStartMoreLcp(Signal* signal, Uint32 nodeId)
     //-------------------------------------------------------------------
 
     sendLCP_FRAG_ORD(signal, nodePtr.p->startedChkpt[startIndex]);
+    return;
   }
 
   /* ----------------------------------------------------------------------- */
-  // When there are no more outstanding LCP reports and there are no one queued
-  // in at least one node, then we are ready to make sure all nodes have at
-  // least two outstanding LCP requests per node and at least two queued for
-  // sending.
+  // If this node has no checkpoints queued up, then attempt to re-fill the
+  // queues across all nodes.
+  // The search for next replicas can be expensive, so we only do it when
+  // the queues are empty.
   /* ----------------------------------------------------------------------- */
   startNextChkpt(signal);
 }//Dbdih::checkStartMoreLcp()