Add msg import info to collatedBlock session stats

Author: SpyCheese

crypto/block/block.cpp

@@ -768,6 +768,15 @@ bool BlockLimitStatus::would_fit(unsigned cls, ton::LogicalTime end_lt, td::uint
                                             limits.collated_data.fits(cls, collated_data_size_estimate));
 }
 
+double BlockLimitStatus::load_fraction(unsigned cls) const {
+  if (cls >= ParamLimits::limits_cnt) {
+    return 0.0;
+  }
+  return std::max({(double)estimate_block_size() / (double)limits.bytes.limit(cls),
+                   (double)gas_used / (double)limits.gas.limit(cls),
+                   (double)collated_data_size_estimate / (double)limits.collated_data.limit(cls)});
+}
+
 // SETS: account_dict, shard_libraries_, mc_state_extra
 //    total_balance{,_extra}, total_validator_fees
 // SETS: out_msg_queue, processed_upto_, ihr_pending (via unpack_out_msg_queue_info)

crypto/block/block.h

@@ -242,6 +242,9 @@ struct ParamLimits {
   td::uint32 hard() const {
     return limits_[3];
   }
+  td::uint32 limit(unsigned cls) const {
+    return limits_[cls];
+  }
   bool compute_medium_limit() {
     limits_[2] = soft() + ((hard() - soft()) >> 1);
     return true;
@@ -299,6 +302,7 @@ struct BlockLimitStatus {
   bool fits(unsigned cls) const;
   bool would_fit(unsigned cls, ton::LogicalTime end_lt, td::uint64 more_gas,
                  const vm::NewCellStorageStat::Stat* extra = nullptr) const;
+  double load_fraction(unsigned cls) const;
   bool add_cell(Ref<vm::Cell> cell) {
     st_stat.add_cell(std::move(cell));
     return true;

crypto/smartcont/CreateState.fif

@@ -179,8 +179,16 @@ variable special-dict
 // bytes-limits gas-limits lt-limits -- c
 { <b x{5d} s, 3 roll param_limits, rot param_limits, swap param_limits, b>
 } : make-block-limits
+// bytes-limits gas-limits lt-limits collated-data-limits imported_queue_bytes imported_queue_msgs -- c
+{ <b x{5e} s,
+  6 roll param_limits, 5 roll param_limits, 4 roll param_limits, 3 roll param_limits,
+  x{d3} s, 2 roll 32 u, 1 roll 32 u, b>
+} : make-block-limits-v2
+
 { make-block-limits 22 config! } : config.mc_block_limits!
 { make-block-limits 23 config! } : config.block_limits!
+{ make-block-limits-v2 22 config! } : config.mc_block_limits_v2!
+{ make-block-limits-v2 23 config! } : config.block_limits_v2!
 
 // mc-block-create-fee bc-block-create-fee
 { <b x{6b} s, rot Gram, swap Gram, b> } : make-block-create-fees

tl/generate/scheme/ton_api.tl

@@ -933,11 +933,18 @@ validatorStats.stats flags:#
 validatorStats.blockLimitsStatus
     bytes:int gas:int lt_delta:int collated_data_bytes:int
     cat_bytes:int cat_gas:int cat_lt_delta:int cat_collated_data_bytes:int
+    load_fraction_queue_cleanup:double load_fraction_dispatch:double
+    load_fraction_internals:double load_fraction_externals:double
+    load_fraction_new_msgs:double
     limits_log:string = validatorStats.BlockLimitsStatus;
-validatorStats.extMsgsStats
+validatorStats.blockStats.extMsgsStats
     ext_msgs_total:int ext_msgs_filtered:int ext_msgs_accepted:int ext_msgs_rejected:int = validatorStats.ExtMsgsStats;
+validatorStats.blockStats.neighborStats shard:tonNode.shardId is_trivial:Bool is_local:Bool msg_limit:int
+    processed_msgs:int skipped_msgs:int limit_reached:Bool = validatorStats.blockStats.NeighborStats;
 validatorStats.blockStats
-    ext_msgs:validatorStats.extMsgsStats transactions:int shard_configuration:(vector tonNode.blockIdExt) = validatorStats.BlockStats;
+    ext_msgs:validatorStats.blockStats.extMsgsStats transactions:int shard_configuration:(vector tonNode.blockIdExt)
+    old_out_msg_queue_size:long new_out_msg_queue_size:long msg_queue_cleaned:int
+    neighbors:(vector validatorStats.blockStats.neighborStats) = validatorStats.BlockStats;
 
 validatorStats.collatedBlock
     block_id:tonNode.blockIdExt collated_data_hash:int256 cc_seqno:int collated_at:double

tl/generate/scheme/ton_api.tlo

@@ -335,8 +335,7 @@ class Collator final : public td::actor::Actor {
   int process_one_new_message(block::NewOutMsg msg, bool enqueue_only = false, Ref<vm::Cell>* is_special = nullptr);
   bool process_inbound_internal_messages();
   bool precheck_inbound_message(Ref<vm::CellSlice> msg, ton::LogicalTime lt);
-  bool process_inbound_message(Ref<vm::CellSlice> msg, ton::LogicalTime lt, td::ConstBitPtr key,
-                               const block::McShardDescr& src_nb);
+  bool process_inbound_message(Ref<vm::CellSlice> msg, ton::LogicalTime lt, td::ConstBitPtr key, int src_nb_idx);
   bool process_inbound_external_messages();
   int process_external_message(Ref<vm::Cell> msg);
   bool process_dispatch_queue();

validator/impl/collator-impl.h

@@ -919,6 +919,7 @@ void Collator::got_neighbor_msg_queues(td::Result<std::map<BlockIdExt, Ref<OutMs
   LOG(INFO) << "neighbor output queues fetched, took " << duration << "s";
   auto res = R.move_as_ok();
   unsigned i = 0;
+  stats_.neighbors.resize(neighbors_.size());
   for (block::McShardDescr& descr : neighbors_) {
     LOG(DEBUG) << "neighbor #" << i << " : " << descr.blk_.to_str();
     auto it = res.find(descr.blk_);
@@ -937,6 +938,13 @@ void Collator::got_neighbor_msg_queue(unsigned i, Ref<OutMsgQueueProof> res) {
   if (res->block_state_proof_.not_null() && !block_id.is_masterchain()) {
     block_state_proofs_.emplace(block_id.root_hash, res->block_state_proof_);
   }
+
+  auto &neighbor_stats = stats_.neighbors.at(i);
+  neighbor_stats.shard = block_id.shard_full();
+  neighbor_stats.is_trivial = shard_intersects(block_id.shard_full(), shard_);
+  neighbor_stats.is_local = res->is_local_;
+  neighbor_stats.msg_limit = res->msg_count_;
+
   Ref<vm::Cell> state_root;
   if (block_id.is_masterchain()) {
     state_root = res->state_root_;
@@ -1000,18 +1008,16 @@ void Collator::got_neighbor_msg_queue(unsigned i, Ref<OutMsgQueueProof> res) {
     return;
   }
   outq_descr.clear();
-  do {
-    // require masterchain blocks referred to in ProcessedUpto
-    // TODO: perform this only if there are messages for this shard in our output queue
-    // .. (have to check the above condition and perform a `break` here) ..
-    // ..
-    for (const auto& entry : descr.processed_upto->list) {
-      Ref<MasterchainStateQ> state;
-      if (!request_aux_mc_state(entry.mc_seqno, state)) {
-        return;
-      }
+  // require masterchain blocks referred to in ProcessedUpto
+  // TODO: perform this only if there are messages for this shard in our output queue
+  // .. (have to check the above condition and perform a `break` here) ..
+  // ..
+  for (const auto& entry : descr.processed_upto->list) {
+    Ref<MasterchainStateQ> state;
+    if (!request_aux_mc_state(entry.mc_seqno, state)) {
+      return;
     }
-  } while (false);
+  }
 }
 
 /**
@@ -1940,7 +1946,7 @@ bool Collator::try_collate() {
   last_proc_int_msg_.second.set_zero();
   first_unproc_int_msg_.first = ~0ULL;
   first_unproc_int_msg_.second.set_ones();
-  old_out_msg_queue_size_ = out_msg_queue_size_;
+  stats_.old_out_msg_queue_size = old_out_msg_queue_size_ = out_msg_queue_size_;
   if (is_masterchain()) {
     LOG(DEBUG) << "getting the list of special smart contracts";
     auto res = config_->get_special_smartcontracts();
@@ -2407,6 +2413,9 @@ bool Collator::dequeue_message(Ref<vm::Cell> msg_envelope, ton::LogicalTime deli
  * @returns True if the cleanup operation was successful, false otherwise.
  */
 bool Collator::out_msg_queue_cleanup() {
+  SCOPE_EXIT {
+    stats_.load_fraction_queue_cleanup = block_limit_status_->load_fraction(block::ParamLimits::cl_normal);
+  };
   LOG(INFO) << "cleaning outbound queue from messages already imported by neighbors";
   if (verbosity >= 2) {
     FLOG(INFO) {
@@ -2492,6 +2501,7 @@ bool Collator::out_msg_queue_cleanup() {
       }
       return true;
     });
+    stats_.msg_queue_cleaned = deleted;
     LOG(WARNING) << "deleted " << deleted << " messages from out_msg_queue after merge, remaining queue size is "
                  << out_msg_queue_size_;
     if (!ok) {
@@ -2572,6 +2582,7 @@ bool Collator::out_msg_queue_cleanup() {
       std::swap(queue_parts[i], queue_parts.back());
       queue_parts.pop_back();
     }
+    stats_.msg_queue_cleaned = deleted;
     LOG(WARNING) << "deleted " << deleted << " messages from out_msg_queue, remaining queue size is "
                  << out_msg_queue_size_;
   }
@@ -3712,12 +3723,13 @@ bool Collator::precheck_inbound_message(Ref<vm::CellSlice> enq_msg, ton::Logical
  * @param enq_msg The inbound message serialized using EnqueuedMsg TLB-scheme.
  * @param lt The logical time of the message.
  * @param key The 32+64+256-bit key of the message.
- * @param src_nb The description of the source neighbor shard.
+ * @param src_nb_idx The index of the source neighbor shard.
  *
  * @returns True if the message was processed successfully, false otherwise.
  */
 bool Collator::process_inbound_message(Ref<vm::CellSlice> enq_msg, ton::LogicalTime lt, td::ConstBitPtr key,
-                                       const block::McShardDescr& src_nb) {
+                                       int src_nb_idx) {
+  const auto& src_nb = neighbors_.at(src_nb_idx);
   ton::LogicalTime enqueued_lt = enq_msg->prefetch_ulong(64);
   auto msg_env = enq_msg->prefetch_ref();
   // 1. unpack MsgEnvelope
@@ -3815,6 +3827,8 @@ bool Collator::process_inbound_message(Ref<vm::CellSlice> enq_msg, ton::LogicalT
                << " enqueued_lt=" << enq_msg_descr.enqueued_lt_ << " has been already processed by us before, skipping";
     // should we dequeue the message if it is ours (after a merge?)
     // (it should have been dequeued by out_msg_queue_cleanup() before)
+    auto &neighbor_stats = stats_.neighbors.at(src_nb_idx);
+    ++neighbor_stats.skipped_msgs;
     return true;
   }
   // 6.1. check whether we have already processed this message by IHR
@@ -3901,12 +3915,17 @@ static std::string block_full_comment(const block::BlockLimitStatus& block_limit
  * @returns True if the processing was successful, false otherwise.
  */
 bool Collator::process_inbound_internal_messages() {
+  SCOPE_EXIT {
+    stats_.load_fraction_internals = block_limit_status_->load_fraction(block::ParamLimits::cl_normal);
+  };
   while (!nb_out_msgs_->is_eof()) {
     block_full_ = !block_limit_status_->fits(block::ParamLimits::cl_normal);
     auto kv = nb_out_msgs_->extract_cur();
     CHECK(kv && kv->msg.not_null());
+    auto &neighbor_stats = stats_.neighbors.at(kv->source);
     if (kv->limit_exceeded) {
       LOG(INFO) << "limit for imported messages is reached, stop processing inbound internal messages";
+      neighbor_stats.limit_reached = true;
       block::EnqueuedMsgDescr enq;
       enq.unpack(kv->msg.write());  // Visit cells to include it in proof
       break;
@@ -3941,13 +3960,14 @@ bool Collator::process_inbound_internal_messages() {
     }
     LOG(DEBUG) << "processing inbound message with (lt,hash)=(" << kv->lt << "," << kv->key.to_hex()
                << ") from neighbor #" << kv->source;
+    ++neighbor_stats.processed_msgs;
     if (verbosity > 2) {
       FLOG(INFO) {
         sb << "inbound message: lt=" << kv->lt << " from=" << kv->source << " key=" << kv->key.to_hex() << " msg=";
         block::gen::t_EnqueuedMsg.print(sb, kv->msg);
       };
     }
-    if (!process_inbound_message(kv->msg, kv->lt, kv->key.cbits(), neighbors_.at(kv->source))) {
+    if (!process_inbound_message(kv->msg, kv->lt, kv->key.cbits(), kv->source)) {
       if (verbosity > 1) {
         FLOG(INFO) {
           sb << "invalid inbound message: lt=" << kv->lt << " from=" << kv->source << " key=" << kv->key.to_hex()
@@ -3970,6 +3990,9 @@ bool Collator::process_inbound_internal_messages() {
  * @returns True if the processing was successful, false otherwise.
  */
 bool Collator::process_inbound_external_messages() {
+  SCOPE_EXIT {
+    stats_.load_fraction_externals = block_limit_status_->load_fraction(block::ParamLimits::cl_soft);
+  };
   if (skip_extmsg_) {
     LOG(INFO) << "skipping processing of inbound external messages";
     return true;
@@ -4089,6 +4112,9 @@ int Collator::process_external_message(Ref<vm::Cell> msg) {
  * @returns True if the processing was successful, false otherwise.
  */
 bool Collator::process_dispatch_queue() {
+  SCOPE_EXIT {
+    stats_.load_fraction_dispatch = block_limit_status_->load_fraction(block::ParamLimits::cl_normal);
+  };
   if (out_msg_queue_size_ > defer_out_queue_size_limit_ && old_out_msg_queue_size_ > hard_defer_out_queue_size_limit_) {
     return true;
   }
@@ -4548,6 +4574,9 @@ bool Collator::enqueue_message(block::NewOutMsg msg, td::RefInt256 fwd_fees_rema
  * @returns True if all new messages were processed successfully, false otherwise.
  */
 bool Collator::process_new_messages(bool enqueue_only) {
+  SCOPE_EXIT {
+    stats_.load_fraction_new_msgs = block_limit_status_->load_fraction(block::ParamLimits::cl_normal);
+  };
   while (!new_msgs.empty()) {
     block::NewOutMsg msg = new_msgs.top();
     new_msgs.pop();
@@ -6380,6 +6409,7 @@ void Collator::finalize_stats() {
       return 0;
     });
   }
+  stats_.new_out_msg_queue_size = out_msg_queue_size_;
 }
 
 /**

validator/impl/collator.cpp

@@ -196,7 +196,7 @@ td::Result<std::vector<td::Ref<OutMsgQueueProof>>> OutMsgQueueProof::fetch(Shard
       if (state_root->get_hash().as_slice() != state_root_hash.as_slice()) {
         return td::Status::Error("state root hash mismatch");
       }
-      res.emplace_back(true, blocks[i], state_root, block_state_proof, f.msg_counts_[i]);
+      res.emplace_back(true, blocks[i], state_root, block_state_proof, false, f.msg_counts_[i]);
 
       data[i].first = blocks[i];
       TRY_RESULT(state, ShardStateQ::fetch(blocks[i], {}, state_root));
@@ -346,7 +346,7 @@ void OutMsgQueueImporter::get_proof_local(std::shared_ptr<CacheEntry> entry, Blo
         auto state = R.move_as_ok();
         if (block.seqno() == 0) {
           std::vector<td::Ref<OutMsgQueueProof>> proof = {
-              td::Ref<OutMsgQueueProof>(true, block, state->root_cell(), td::Ref<vm::Cell>{})};
+              td::Ref<OutMsgQueueProof>(true, block, state->root_cell(), td::Ref<vm::Cell>{}, true)};
           td::actor::send_closure(SelfId, &OutMsgQueueImporter::got_proof, entry, std::move(proof), ProofSource::Local);
           return;
         }
@@ -362,7 +362,7 @@ void OutMsgQueueImporter::get_proof_local(std::shared_ptr<CacheEntry> entry, Blo
               }
               Ref<vm::Cell> block_state_proof = create_block_state_proof(R.ok()->root_cell()).move_as_ok();
               std::vector<td::Ref<OutMsgQueueProof>> proof = {
-                  td::Ref<OutMsgQueueProof>(true, block, state->root_cell(), std::move(block_state_proof))};
+                  td::Ref<OutMsgQueueProof>(true, block, state->root_cell(), std::move(block_state_proof), true)};
               td::actor::send_closure(SelfId, &OutMsgQueueImporter::got_proof, entry, std::move(proof),
                                       ProofSource::Local);
             });

validator/impl/out-msg-queue-proof.cpp

@@ -26,17 +26,19 @@ namespace validator {
 using td::Ref;
 
 struct OutMsgQueueProof : public td::CntObject {
-  OutMsgQueueProof(BlockIdExt block_id, Ref<vm::Cell> state_root, Ref<vm::Cell> block_state_proof,
+  OutMsgQueueProof(BlockIdExt block_id, Ref<vm::Cell> state_root, Ref<vm::Cell> block_state_proof, bool is_local,
                    td::int32 msg_count = -1)
       : block_id_(block_id)
       , state_root_(std::move(state_root))
       , block_state_proof_(std::move(block_state_proof))
+      , is_local_(is_local)
       , msg_count_(msg_count) {
   }
 
   BlockIdExt block_id_;
   Ref<vm::Cell> state_root_;
   Ref<vm::Cell> block_state_proof_;
+  bool is_local_ = false;
   td::int32 msg_count_;  // -1 - no limit
 
   static td::Result<std::vector<td::Ref<OutMsgQueueProof>>> fetch(ShardIdFull dst_shard, std::vector<BlockIdExt> blocks,

validator/interfaces/out-msg-queue-proof.h

@@ -69,31 +69,63 @@ struct CollationStats {
   td::uint32 estimated_bytes = 0, gas = 0, lt_delta = 0, estimated_collated_data_bytes = 0;
   int cat_bytes = 0, cat_gas = 0, cat_lt_delta = 0, cat_collated_data_bytes = 0;
   std::string limits_log;
+  double total_time = 0.0, work_time = 0.0, cpu_work_time = 0.0;
+  std::string time_stats;
+
   td::uint32 transactions = 0;
   std::vector<BlockIdExt> shard_configuration;
   td::uint32 ext_msgs_total = 0;
   td::uint32 ext_msgs_filtered = 0;
   td::uint32 ext_msgs_accepted = 0;
   td::uint32 ext_msgs_rejected = 0;
-  double total_time = 0.0, work_time = 0.0, cpu_work_time = 0.0;
-  std::string time_stats;
+
+  td::uint64 old_out_msg_queue_size = 0;
+  td::uint64 new_out_msg_queue_size = 0;
+  td::uint32 msg_queue_cleaned = 0;
+  struct NeighborStats {
+    ShardIdFull shard;
+    bool is_trivial = false;
+    bool is_local = false;
+    int msg_limit = -1;
+    td::uint32 processed_msgs = 0;
+    td::uint32 skipped_msgs = 0;
+    bool limit_reached = false;
+
+    tl_object_ptr<ton_api::validatorStats_blockStats_neighborStats> tl() const {
+      return create_tl_object<ton_api::validatorStats_blockStats_neighborStats>(
+          create_tl_shard_id(shard), is_trivial, is_local, msg_limit, processed_msgs, skipped_msgs, limit_reached);
+    }
+  };
+  std::vector<NeighborStats> neighbors;
+
+  double load_fraction_queue_cleanup = -1.0;
+  double load_fraction_dispatch = -1.0;
+  double load_fraction_internals = -1.0;
+  double load_fraction_externals = -1.0;
+  double load_fraction_new_msgs = -1.0;
 
   tl_object_ptr<ton_api::validatorStats_collatedBlock> tl() const {
-    std::vector<tl_object_ptr<ton_api::tonNode_blockIdExt>> shards;
+    std::vector<tl_object_ptr<ton_api::tonNode_blockIdExt>> shards_obj;
     for (const BlockIdExt& block_id : shard_configuration) {
-      shards.push_back(create_tl_block_id(block_id));
+      shards_obj.push_back(create_tl_block_id(block_id));
+    }
+    std::vector<tl_object_ptr<ton_api::validatorStats_blockStats_neighborStats>> neighbors_obj;
+    for (const NeighborStats& neighbor : neighbors) {
+      neighbors_obj.push_back(neighbor.tl());
     }
     auto block_stats = create_tl_object<ton_api::validatorStats_blockStats>(
-        create_tl_object<ton_api::validatorStats_extMsgsStats>(ext_msgs_total, ext_msgs_filtered, ext_msgs_accepted,
-                                                               ext_msgs_rejected),
-        transactions, std::move(shards));
+        create_tl_object<ton_api::validatorStats_blockStats_extMsgsStats>(ext_msgs_total, ext_msgs_filtered,
+                                                                          ext_msgs_accepted, ext_msgs_rejected),
+        transactions, std::move(shards_obj), old_out_msg_queue_size, new_out_msg_queue_size, msg_queue_cleaned,
+        std::move(neighbors_obj));
     return create_tl_object<ton_api::validatorStats_collatedBlock>(
         create_tl_block_id(block_id), collated_data_hash, cc_seqno, collated_at, actual_bytes,
         actual_collated_data_bytes, attempt, self.bits256_value(), is_validator, total_time, work_time, cpu_work_time,
         time_stats,
-        create_tl_object<ton_api::validatorStats_blockLimitsStatus>(estimated_bytes, gas, lt_delta,
-                                                                    estimated_collated_data_bytes, cat_bytes, cat_gas,
-                                                                    cat_lt_delta, cat_collated_data_bytes, limits_log),
+        create_tl_object<ton_api::validatorStats_blockLimitsStatus>(
+            estimated_bytes, gas, lt_delta, estimated_collated_data_bytes, cat_bytes, cat_gas, cat_lt_delta,
+            cat_collated_data_bytes, load_fraction_queue_cleanup, load_fraction_dispatch, load_fraction_internals,
+            load_fraction_externals, load_fraction_new_msgs, limits_log),
         std::move(block_stats));
   }
 };