Bug#36579842 Backport ndbinfo.transporter_details enhancements to 8.0

Part 1/2, backport of commit :
    7357e3e5decb7128bf703e8b30c31b75c235f4c9

    WL#7662 Sendbuffer usage monitoring

    The NdbInfo transporter_details table is extended with
    columns giving per transporter sendbuffer usage
    information for SendBuffers in the data nodes :

    New columns are :
     - sendbuffer_used_bytes
       Number of bytes of signal data stored in SendBuffer
       pending send using this transporter.

     - sendbuffer_max_used_bytes
       Maximum number of bytes of signal data stored in SendBuffer
       pending send since most recent transporter connect

     - sendbuffer_alloc_bytes
       Number of bytes of SendBuffer allocated for storing signal
       data pending send using this transporter.
       SendBuffer is allocated in pages which may not be fully
       utilised

     - sendbuffer_max_alloc_bytes
       Maximum number of bytes of SendBuffer allocated for pending
       sends since most recent transporter connect

    The values are sampled before and after sending.

Change-Id: I7ef4f6f9dfec982b365c6345311c62d163ccc20e

Author: frazerclement

mysql-test/suite/ndb/r/ndbinfo_plans.result

@@ -62,7 +62,7 @@ index_columns	20	80
 ndb$acc_operations	15	64
 ndb$backup_id	1	20
 ndb$blocks	29	20
-ndb$columns	550	44
+ndb$columns	554	44
 ndb$config_nodes	34	28
 ndb$config_params	167	120
 ndb$config_values	330	24
@@ -107,7 +107,7 @@ ndb$threadblocks	124	16
 ndb$threads	26	40
 ndb$threadstat	22	144
 ndb$transactions	5	44
-ndb$transporter_details	32	76
+ndb$transporter_details	32	108
 ndb$transporters	32	64
 
 ## List the tables where estimated size equals actual size.

storage/ndb/include/transporter/TransporterRegistry.hpp

@@ -430,6 +430,12 @@ class TransporterRegistry {
    */
   Uint32 get_connect_count(TrpId trpId) const;
 
+  /**
+   * Update send buffer allocated and used bytes for given transporter
+   */
+  void update_send_buffer_usage(TrpId trpId, Uint64 allocBytes,
+                                Uint64 usedBytes);
+
   /**
    * Set or clear overloaded bit.
    * Query if any overloaded bit is set.
@@ -454,6 +460,14 @@ class TransporterRegistry {
    */
   Uint32 get_slowdown_count(NodeId nodeId) const;
 
+  /**
+   * Get SendBuffer alloc + usage info
+   */
+  Uint64 get_send_buffer_alloc_bytes(TrpId trpId) const;
+  Uint64 get_send_buffer_used_bytes(TrpId trpId) const;
+  Uint64 get_send_buffer_max_alloc_bytes(TrpId trpId) const;
+  Uint64 get_send_buffer_max_used_bytes(TrpId trpId) const;
+
   /**
    * prepareSend
    *

storage/ndb/plugin/ha_ndbinfo_sql.cc

@@ -660,7 +660,9 @@ static struct view {
      " END AS status, "
      " remote_address, bytes_sent, bytes_received, "
      " connect_count, "
-     " overloaded, overload_count, slowdown, slowdown_count, encrypted "
+     " overloaded, overload_count, slowdown, slowdown_count, encrypted, "
+     " sendbuffer_used_bytes, sendbuffer_max_used_bytes, "
+     " sendbuffer_alloc_bytes, sendbuffer_max_alloc_bytes "
      "FROM `ndbinfo`.`ndb$transporter_details`"},
     {"ndbinfo", "transporters",
      "SELECT node_id, remote_node_id, "

storage/ndb/src/common/debugger/NdbinfoTables.cpp

@@ -105,8 +105,8 @@ DECLARE_NDBINFO_TABLE(POOLS, 14) = {
      {"resource_id", Ndbinfo::Number, ""},
      {"type_id", Ndbinfo::Number, "Record type id within resource"}}};
 
-DECLARE_NDBINFO_TABLE(TRANSPORTER_DETAILS, 14) = {
-    {"transporter_details", 14, 0,
+DECLARE_NDBINFO_TABLE(TRANSPORTER_DETAILS, 18) = {
+    {"transporter_details", 18, 0,
      [](const Ndbinfo::Counts &counts) {
        return (counts.data_nodes) * (counts.all_nodes - 1);
      },
@@ -131,7 +131,15 @@ DECLARE_NDBINFO_TABLE(TRANSPORTER_DETAILS, 14) = {
      {"slowdown", Ndbinfo::Number, "Is link requesting slowdown"},
      {"slowdown_count", Ndbinfo::Number,
       "Number of slowdown onsets since connect"},
-     {"encrypted", Ndbinfo::Number, "Is link using TLS encryption"}}};
+     {"encrypted", Ndbinfo::Number, "Is link using TLS encryption"},
+
+     {"sendbuffer_used_bytes", Ndbinfo::Number64, "SendBuffer bytes in use"},
+     {"sendbuffer_max_used_bytes", Ndbinfo::Number64,
+      "SendBuffer historical max bytes in use"},
+     {"sendbuffer_alloc_bytes", Ndbinfo::Number64,
+      "SendBuffer bytes allocated"},
+     {"sendbuffer_max_alloc_bytes", Ndbinfo::Number64,
+      "SendBuffer historical max bytes allocated"}}};
 
 DECLARE_NDBINFO_TABLE(TRANSPORTERS, 11) = {
     {"transporters", 11, 0,

storage/ndb/src/common/transporter/Transporter.cpp

@@ -83,6 +83,10 @@ Transporter::Transporter(TransporterRegistry &t_reg, TrpId transporter_index,
       m_connect_count(0),
       m_overload_count(0),
       m_slowdown_count(0),
+      m_send_buffer_alloc_bytes(0),
+      m_send_buffer_max_alloc_bytes(0),
+      m_send_buffer_used_bytes(0),
+      m_send_buffer_max_used_bytes(0),
       isMgmConnection(_isMgmConnection),
       m_connected(false),
       m_type(_type),
@@ -463,6 +467,10 @@ void Transporter::resetCounters() {
   m_bytes_received = 0;
   m_overload_count = 0;
   m_slowdown_count = 0;
+  m_send_buffer_alloc_bytes = 0;
+  m_send_buffer_max_alloc_bytes = 0;
+  m_send_buffer_used_bytes = 0;
+  m_send_buffer_max_used_bytes = 0;
 }
 
 void Transporter::checksum_state::dumpBadChecksumInfo(

storage/ndb/src/common/transporter/Transporter.hpp

@@ -151,6 +151,23 @@ class Transporter {
                                                used >= m_slowdown_limit);
   }
 
+  /**
+   * Update ndbinfo statistics about sendbuffer bytes allocated and
+   * used by transporter.
+   * Note that allocated bytes may be sparsely populated pages,
+   * resulting in an overallocation of pages as well.
+   * We expect the buffers soon to be packed in such cases.
+   */
+  void update_send_buffer_usage(Uint64 allocBytes, Uint64 usedBytes) {
+    m_send_buffer_alloc_bytes = allocBytes;
+    if (allocBytes > m_send_buffer_max_alloc_bytes)
+      m_send_buffer_max_alloc_bytes = allocBytes;
+
+    m_send_buffer_used_bytes = usedBytes;
+    if (usedBytes > m_send_buffer_max_used_bytes)
+      m_send_buffer_max_used_bytes = usedBytes;
+  }
+
   virtual bool doSend(bool need_wakeup = true) = 0;
 
   /* Get the configured maximum send buffer usage. */
@@ -171,6 +188,12 @@ class Transporter {
 
   TransporterType getTransporterType() const;
 
+  Uint64 get_alloc_bytes() const { return m_send_buffer_alloc_bytes; }
+  Uint64 get_max_alloc_bytes() const { return m_send_buffer_max_alloc_bytes; }
+
+  Uint64 get_used_bytes() const { return m_send_buffer_used_bytes; }
+  Uint64 get_max_used_bytes() const { return m_send_buffer_max_used_bytes; }
+
  protected:
   Transporter(TransporterRegistry &, TrpId transporter_index, TransporterType,
               const char *lHostName, const char *rHostName, int s_port,
@@ -228,13 +251,20 @@ class Transporter {
   /* Overload limit, as configured with the OverloadLimit config parameter. */
   Uint32 m_overload_limit;
   Uint32 m_slowdown_limit;
-  void resetCounters();
   Uint64 m_bytes_sent;
   Uint64 m_bytes_received;
   Uint32 m_connect_count;
   Uint32 m_overload_count;
   Uint32 m_slowdown_count;
 
+  Uint64 m_send_buffer_alloc_bytes;
+  Uint64 m_send_buffer_max_alloc_bytes;  // Historic max use
+
+  Uint64 m_send_buffer_used_bytes;
+  Uint64 m_send_buffer_max_used_bytes;  // Historic max use
+
+  void resetCounters();
+
   // Sending/Receiving socket used by both client and server
   NdbSocket theSocket;
 
@@ -361,9 +391,14 @@ inline Uint32 Transporter::fetch_send_iovec_data(struct iovec dst[],
 }
 
 inline void Transporter::iovec_data_sent(int nBytesSent) {
-  Uint32 used_bytes =
+  Uint32 remaining_bytes =
       get_callback_obj()->bytes_sent(m_transporter_index, nBytesSent);
-  update_status_overloaded(used_bytes);
+  update_status_overloaded(remaining_bytes);
+
+  if (remaining_bytes == 0) {
+    m_send_buffer_alloc_bytes = 0;
+    m_send_buffer_used_bytes = 0;
+  }
 }
 
 inline bool Transporter::checksum_state::compute(const void *buf, size_t len) {

storage/ndb/src/common/transporter/TransporterRegistry.cpp

@@ -3283,6 +3283,18 @@ void TransporterRegistry::inc_overload_count(NodeId nodeId) {
   theNodeIdTransporters[nodeId]->inc_overload_count();
 }
 
+/**
+ * TR need to inform Transporter about how much pending buffered
+ * send data there is.
+ */
+void TransporterRegistry::update_send_buffer_usage(TrpId trpId,
+                                                   Uint64 allocBytes,
+                                                   Uint64 usedBytes) {
+  assert(trpId < maxTransporters);
+  assert(allTransporters[trpId] != nullptr);
+  allTransporters[trpId]->update_send_buffer_usage(allocBytes, usedBytes);
+}
+
 void TransporterRegistry::inc_slowdown_count(NodeId nodeId) {
   assert(nodeId < MAX_NODES);
   assert(theNodeIdTransporters[nodeId] != nullptr);
@@ -3307,6 +3319,30 @@ Uint32 TransporterRegistry::get_connect_count(TrpId trpId) const {
   return allTransporters[trpId]->get_connect_count();
 }
 
+Uint64 TransporterRegistry::get_send_buffer_alloc_bytes(TrpId trpId) const {
+  assert(trpId < MAX_NTRANSPORTERS);
+  assert(allTransporters[trpId] != nullptr);
+  return allTransporters[trpId]->get_alloc_bytes();
+}
+
+Uint64 TransporterRegistry::get_send_buffer_max_alloc_bytes(TrpId trpId) const {
+  assert(trpId < MAX_NTRANSPORTERS);
+  assert(allTransporters[trpId] != nullptr);
+  return allTransporters[trpId]->get_max_alloc_bytes();
+}
+
+Uint64 TransporterRegistry::get_send_buffer_used_bytes(TrpId trpId) const {
+  assert(trpId < MAX_NTRANSPORTERS);
+  assert(allTransporters[trpId] != nullptr);
+  return allTransporters[trpId]->get_used_bytes();
+}
+
+Uint64 TransporterRegistry::get_send_buffer_max_used_bytes(TrpId trpId) const {
+  assert(trpId < MAX_NTRANSPORTERS);
+  assert(allTransporters[trpId] != nullptr);
+  return allTransporters[trpId]->get_max_used_bytes();
+}
+
 void TransporterRegistry::get_trps_for_node(NodeId nodeId, TrpId *trp_ids,
                                             Uint32 &num_ids,
                                             Uint32 max_size) const {

storage/ndb/src/kernel/blocks/trpman.cpp

@@ -503,7 +503,6 @@ void Trpman::execDBINFO_SCANREQ(Signal *signal) {
 
         /* Connect count, overload and slowdown states */
         row.write_uint32(globalTransporterRegistry.get_connect_count(trpId));
-
         /* FIXME: overload & slowdown is still pr NodeId */
         row.write_uint32(
             globalTransporterRegistry.get_status_overloaded().get(nodeId));
@@ -515,6 +514,16 @@ void Trpman::execDBINFO_SCANREQ(Signal *signal) {
         /* TLS, not available until 8.3 -> is_encrypted=false*/
         row.write_uint32(false);
 
+        /* Send buffer bytes/pages usage */
+        row.write_uint64(
+            globalTransporterRegistry.get_send_buffer_used_bytes(trpId));
+        row.write_uint64(
+            globalTransporterRegistry.get_send_buffer_max_used_bytes(trpId));
+        row.write_uint64(
+            globalTransporterRegistry.get_send_buffer_alloc_bytes(trpId));
+        row.write_uint64(
+            globalTransporterRegistry.get_send_buffer_max_alloc_bytes(trpId));
+
         ndbinfo_send_row(signal, req, row, rl);
         trpId++;
         if (rl.need_break(req)) {

storage/ndb/src/kernel/vm/TransporterCallback.cpp

@@ -647,14 +647,23 @@ Uint32 TransporterCallbackKernelNonMT::get_bytes_to_send_iovec(
 
   Uint32 count = 0;
   SendBufferPage *page = b->m_first_page;
-  while (page != NULL && count < max) {
+  while (page != nullptr && count < max) {
     dst[count].iov_base = page->m_data + page->m_start;
     dst[count].iov_len = page->m_bytes;
     assert(page->m_start + page->m_bytes <= page->max_data_bytes());
     page = page->m_next;
     count++;
   }
 
+  // For send_buffer_usage we count pages beyond 'max' as well
+  Uint32 pages = count;
+  while (unlikely(page != nullptr)) {
+    page = page->m_next;
+    pages++;
+  }
+  globalTransporterRegistry.update_send_buffer_usage(
+      trp_id, Uint64(pages) * SendBufferPage::PGSIZE, b->m_used_bytes);
+
   return count;
 }