Merge branch 'accelerator' into accelerator-test

assembly/wasm/fift-func-wasm-build-ubuntu.sh

@@ -85,7 +85,7 @@ cd ..
 if [ ! -f "openssl/openssl_em" ]; then
   cd openssl
   make clean
-  emconfigure ./Configure linux-generic32 no-shared no-dso no-engine no-unit-test
+  emconfigure ./Configure linux-generic32 no-shared no-dso no-engine no-unit-test no-tests no-fuzz-afl no-fuzz-libfuzzer
   sed -i 's/CROSS_COMPILE=.*/CROSS_COMPILE=/g' Makefile
   sed -i 's/-ldl//g' Makefile
   sed -i 's/-O3/-Os/g' Makefile

blockchain-explorer/blockchain-explorer-query.cpp

@@ -1432,7 +1432,7 @@ void HttpQueryStatus::finish_query() {
       for (td::uint32 i = 0; i < results_.ips.size(); i++) {
         A << "<tr>";
         if (results_.ips[i].is_valid()) {
-          A << "<td>" << results_.ips[i] << "</td>";
+          A << "<td>" << results_.ips[i].get_ip_str() << ":" << results_.ips[i].get_port() << "</td>";
         } else {
           A << "<td>hidden</td>";
         }

blockchain-explorer/blockchain-explorer.cpp

@@ -155,8 +155,12 @@ class CoreActor : public CoreActorInterface {
   td::int32 attempt_ = 0;
   td::int32 waiting_ = 0;
 
-  //void run_queries();
-  void got_result(td::uint32 idx, td::int32 attempt, td::Result<td::BufferSlice> data);
+  size_t n_servers_ = 0;
+
+  void run_queries();
+  void got_servers_ready(td::int32 attempt, std::vector<bool> ready);
+  void send_ping(td::uint32 idx);
+  void got_ping_result(td::uint32 idx, td::int32 attempt, td::Result<td::BufferSlice> data);
 
   void add_result() {
     if (new_result_) {
@@ -175,7 +179,7 @@ class CoreActor : public CoreActorInterface {
       add_result();
     }
     attempt_ = t;
-    //run_queries();
+    run_queries();
     alarm_timestamp() = td::Timestamp::at_unix((attempt_ + 1) * 60);
   }
 
@@ -450,15 +454,55 @@ class CoreActor : public CoreActorInterface {
       addrs_.push_back(remote_addr_);
       servers.push_back(liteclient::LiteServerConfig{ton::adnl::AdnlNodeIdFull{remote_public_key_}, remote_addr_});
     }
-    client_ = liteclient::ExtClient::create(std::move(servers), make_callback());
+    n_servers_ = servers.size();
+    client_ = liteclient::ExtClient::create(std::move(servers), make_callback(), true);
     daemon_ = MHD_start_daemon(MHD_USE_SELECT_INTERNALLY, static_cast<td::uint16>(http_port_), nullptr, nullptr,
                                &process_http_request, nullptr, MHD_OPTION_NOTIFY_COMPLETED, request_completed, nullptr,
                                MHD_OPTION_THREAD_POOL_SIZE, 16, MHD_OPTION_END);
     CHECK(daemon_ != nullptr);
   }
 };
 
-void CoreActor::got_result(td::uint32 idx, td::int32 attempt, td::Result<td::BufferSlice> R) {
+void CoreActor::run_queries() {
+  waiting_ = 0;
+  new_result_ = std::make_shared<RemoteNodeStatus>(n_servers_, td::Timestamp::at_unix(attempt_ * 60));
+  td::actor::send_closure(client_, &liteclient::ExtClient::get_servers_status,
+                          [SelfId = actor_id(this), attempt = attempt_](td::Result<std::vector<bool>> R) {
+                            R.ensure();
+                            td::actor::send_closure(SelfId, &CoreActor::got_servers_ready, attempt, R.move_as_ok());
+                          });
+}
+
+void CoreActor::got_servers_ready(td::int32 attempt, std::vector<bool> ready) {
+  if (attempt != attempt_) {
+    return;
+  }
+  CHECK(ready.size() == n_servers_);
+  for (td::uint32 i = 0; i < n_servers_; i++) {
+    if (ready[i]) {
+      send_ping(i);
+    }
+  }
+  CHECK(waiting_ >= 0);
+  if (waiting_ == 0) {
+    add_result();
+  }
+}
+
+void CoreActor::send_ping(td::uint32 idx) {
+  waiting_++;
+  auto query = ton::create_tl_object<ton::lite_api::liteServer_getMasterchainInfo>();
+  auto q = ton::create_tl_object<ton::lite_api::liteServer_query>(serialize_tl_object(query, true));
+
+  auto P =
+      td::PromiseCreator::lambda([SelfId = actor_id(this), idx, attempt = attempt_](td::Result<td::BufferSlice> R) {
+        td::actor::send_closure(SelfId, &CoreActor::got_ping_result, idx, attempt, std::move(R));
+      });
+  td::actor::send_closure(client_, &liteclient::ExtClient::send_query_to_server, "query", serialize_tl_object(q, true),
+                          idx, td::Timestamp::in(10.0), std::move(P));
+}
+
+void CoreActor::got_ping_result(td::uint32 idx, td::int32 attempt, td::Result<td::BufferSlice> R) {
   if (attempt != attempt_) {
     return;
   }
@@ -499,18 +543,6 @@ void CoreActor::got_result(td::uint32 idx, td::int32 attempt, td::Result<td::Buf
   }
 }
 
-/*void CoreActor::run_queries() {
-  waiting_ = 0;
-  new_result_ = std::make_shared<RemoteNodeStatus>(ready_.size(), td::Timestamp::at_unix(attempt_ * 60));
-  for (td::uint32 i = 0; i < ready_.size(); i++) {
-    send_query(i);
-  }
-  CHECK(waiting_ >= 0);
-  if (waiting_ == 0) {
-    add_result();
-  }
-}*/
-
 void CoreActor::send_lite_query(td::BufferSlice query, td::Promise<td::BufferSlice> promise) {
   auto P = td::PromiseCreator::lambda([promise = std::move(promise)](td::Result<td::BufferSlice> R) mutable {
     if (R.is_error()) {

catchain/catchain.h

@@ -96,6 +96,7 @@ class CatChain : public td::actor::Actor {
   virtual void send_query_via(const PublicKeyHash &dst, std::string name, td::Promise<td::BufferSlice> promise,
                               td::Timestamp timeout, td::BufferSlice query, td::uint64 max_answer_size,
                               td::actor::ActorId<adnl::AdnlSenderInterface> via) = 0;
+  virtual void get_source_heights(td::Promise<std::vector<CatChainBlockHeight>> promise) = 0;
   virtual void destroy() = 0;
 
   static td::actor::ActorOwn<CatChain> create(std::unique_ptr<Callback> callback, const CatChainOptions &opts,

catchain/catchain.hpp

@@ -115,6 +115,15 @@ class CatChainImpl : public CatChain {
     td::actor::send_closure(receiver_, &CatChainReceiverInterface::send_custom_query_data_via, dst, name,
                             std::move(promise), timeout, std::move(query), max_answer_size, via);
   }
+  void get_source_heights(td::Promise<std::vector<CatChainBlockHeight>> promise) override {
+    std::vector<CatChainBlockHeight> heights(top_source_blocks_.size(), 0);
+    for (size_t i = 0; i < top_source_blocks_.size(); ++i) {
+      if (top_source_blocks_[i]) {
+        heights[i] = top_source_blocks_[i]->height();
+      }
+    }
+    promise.set_result(std::move(heights));
+  }
   void destroy() override;
   CatChainImpl(std::unique_ptr<Callback> callback, const CatChainOptions &opts,
                td::actor::ActorId<keyring::Keyring> keyring, td::actor::ActorId<adnl::Adnl> adnl,

dht-server/dht-server.cpp

@@ -1233,15 +1233,19 @@ int main(int argc, char *argv[]) {
   });
   td::uint32 threads = 7;
   p.add_checked_option(
-      't', "threads", PSTRING() << "number of threads (default=" << threads << ")", [&](td::Slice fname) {
+      't', "threads", PSTRING() << "number of threads (default=" << threads << ")", [&](td::Slice arg) {
         td::int32 v;
         try {
-          v = std::stoi(fname.str());
+          v = std::stoi(arg.str());
         } catch (...) {
           return td::Status::Error(ton::ErrorCode::error, "bad value for --threads: not a number");
         }
-        if (v < 1 || v > 256) {
-          return td::Status::Error(ton::ErrorCode::error, "bad value for --threads: should be in range [1..256]");
+        if (v <= 0) {
+          return td::Status::Error(ton::ErrorCode::error, "bad value for --threads: should be > 0");
+        }
+        if (v > 127) {
+          LOG(WARNING) << "`--threads " << v << "` is too big, effective value will be 127";
+          v = 127;
         }
         threads = v;
         return td::Status::OK();

lite-client/ext-client.cpp

@@ -22,8 +22,8 @@ namespace liteclient {
 
 class ExtClientImpl : public ExtClient {
  public:
-  ExtClientImpl(std::vector<LiteServerConfig> liteservers, td::unique_ptr<Callback> callback)
-      : callback_(std::move(callback)) {
+  ExtClientImpl(std::vector<LiteServerConfig> liteservers, td::unique_ptr<Callback> callback, bool connect_to_all)
+      : callback_(std::move(callback)), connect_to_all_(connect_to_all) {
     CHECK(!liteservers.empty());
     servers_.resize(liteservers.size());
     for (size_t i = 0; i < servers_.size(); ++i) {
@@ -36,15 +36,65 @@ class ExtClientImpl : public ExtClient {
     LOG(INFO) << "Started ext client, " << servers_.size() << " liteservers";
     td::Random::Fast rnd;
     td::random_shuffle(td::as_mutable_span(servers_), rnd);
+    server_indices_.resize(servers_.size());
+    for (size_t i = 0; i < servers_.size(); ++i) {
+      server_indices_[servers_[i].idx] = i;
+    }
+
+    if (connect_to_all_) {
+      for (size_t i = 0; i < servers_.size(); ++i) {
+        prepare_server(i, nullptr);
+      }
+    }
   }
 
   void send_query(std::string name, td::BufferSlice data, td::Timestamp timeout,
                   td::Promise<td::BufferSlice> promise) override {
     QueryInfo query_info = get_query_info(data);
     TRY_RESULT_PROMISE(promise, server_idx, select_server(query_info));
+    send_query_internal(std::move(name), std::move(data), std::move(query_info), server_idx, timeout,
+                        std::move(promise));
+  }
+
+  void send_query_to_server(std::string name, td::BufferSlice data, size_t server_idx, td::Timestamp timeout,
+                            td::Promise<td::BufferSlice> promise) override {
+    if (server_idx >= servers_.size()) {
+      promise.set_error(td::Status::Error(PSTRING() << "server idx " << server_idx << " is too big"));
+      return;
+    }
+    server_idx = server_indices_[server_idx];
+    QueryInfo query_info = get_query_info(data);
+    prepare_server(server_idx, &query_info);
+    send_query_internal(std::move(name), std::move(data), std::move(query_info), server_idx, timeout,
+                        std::move(promise));
+  }
+
+  void get_servers_status(td::Promise<std::vector<bool>> promise) override {
+    std::vector<bool> status(servers_.size());
+    for (const Server& s : servers_) {
+      status[s.idx] = s.alive;
+    }
+    promise.set_result(std::move(status));
+  }
+
+  void reset_servers() override {
+    LOG(INFO) << "Force resetting all liteservers";
+    for (Server& server : servers_) {
+      server.alive = false;
+      server.timeout = {};
+      server.ignore_until = {};
+      server.client.reset();
+    }
+  }
+
+ private:
+  void send_query_internal(std::string name, td::BufferSlice data, QueryInfo query_info, size_t server_idx,
+                           td::Timestamp timeout, td::Promise<td::BufferSlice> promise) {
     auto& server = servers_[server_idx];
     CHECK(!server.client.empty());
-    alarm_timestamp().relax(server.timeout = td::Timestamp::in(MAX_NO_QUERIES_TIMEOUT));
+    if (!connect_to_all_) {
+      alarm_timestamp().relax(server.timeout = td::Timestamp::in(MAX_NO_QUERIES_TIMEOUT));
+    }
     td::Promise<td::BufferSlice> P = [SelfId = actor_id(this), server_idx,
                                       promise = std::move(promise)](td::Result<td::BufferSlice> R) mutable {
       if (R.is_error() &&
@@ -59,17 +109,6 @@ class ExtClientImpl : public ExtClient {
                  std::move(P));
   }
 
-  void reset_servers() override {
-    LOG(INFO) << "Force resetting all liteservers";
-    for (Server& server : servers_) {
-      server.alive = false;
-      server.timeout = {};
-      server.ignore_until = {};
-      server.client.reset();
-    }
-  }
-
- private:
   td::Result<size_t> select_server(const QueryInfo& query_info) {
     for (size_t i = 0; i < servers_.size(); ++i) {
       if (servers_[i].alive && servers_[i].config.accepts_query(query_info)) {
@@ -101,12 +140,22 @@ class ExtClientImpl : public ExtClient {
     if (server_idx == servers_.size()) {
       return td::Status::Error(PSTRING() << "no liteserver for query " << query_info.to_str());
     }
+    prepare_server(server_idx, &query_info);
+    return server_idx;
+  }
+
+  void prepare_server(size_t server_idx, const QueryInfo* query_info) {
     Server& server = servers_[server_idx];
+    if (server.alive) {
+      return;
+    }
     server.alive = true;
     server.ignore_until = {};
-    alarm_timestamp().relax(server.timeout = td::Timestamp::in(MAX_NO_QUERIES_TIMEOUT));
+    if (!connect_to_all_) {
+      alarm_timestamp().relax(server.timeout = td::Timestamp::in(MAX_NO_QUERIES_TIMEOUT));
+    }
     if (!server.client.empty()) {
-      return server_idx;
+      return;
     }
 
     class Callback : public ton::adnl::AdnlExtClient::Callback {
@@ -124,10 +173,9 @@ class ExtClientImpl : public ExtClient {
       size_t idx_;
     };
     LOG(INFO) << "Connecting to liteserver #" << server.idx << " (" << server.config.addr.get_ip_str() << ":"
-              << server.config.addr.get_port() << ") for query " << query_info.to_str();
+              << server.config.addr.get_port() << ") for query " << (query_info ? query_info->to_str() : "[none]");
     server.client = ton::adnl::AdnlExtClient::create(server.config.adnl_id, server.config.addr,
                                                      std::make_unique<Callback>(actor_id(this), server_idx));
-    return server_idx;
   }
 
   struct Server {
@@ -139,12 +187,17 @@ class ExtClientImpl : public ExtClient {
     td::Timestamp ignore_until = td::Timestamp::never();
   };
   std::vector<Server> servers_;
+  std::vector<size_t> server_indices_;
 
   td::unique_ptr<Callback> callback_;
+  bool connect_to_all_ = false;
   static constexpr double MAX_NO_QUERIES_TIMEOUT = 100.0;
   static constexpr double BAD_SERVER_TIMEOUT = 30.0;
 
   void alarm() override {
+    if (connect_to_all_) {
+      return;
+    }
     for (Server& server : servers_) {
       if (server.timeout && server.timeout.is_in_past()) {
         LOG(INFO) << "Closing connection to liteserver #" << server.idx << " (" << server.config.addr.get_ip_str()
@@ -168,7 +221,8 @@ td::actor::ActorOwn<ExtClient> ExtClient::create(ton::adnl::AdnlNodeIdFull dst,
 }
 
 td::actor::ActorOwn<ExtClient> ExtClient::create(std::vector<LiteServerConfig> liteservers,
-                                                 td::unique_ptr<Callback> callback) {
-  return td::actor::create_actor<ExtClientImpl>("ExtClient", std::move(liteservers), std::move(callback));
+                                                 td::unique_ptr<Callback> callback, bool connect_to_all) {
+  return td::actor::create_actor<ExtClientImpl>("ExtClient", std::move(liteservers), std::move(callback),
+                                                connect_to_all);
 }
 }  // namespace liteclient

lite-client/ext-client.h

@@ -30,12 +30,19 @@ class ExtClient : public td::actor::Actor {
 
   virtual void send_query(std::string name, td::BufferSlice data, td::Timestamp timeout,
                           td::Promise<td::BufferSlice> promise) = 0;
+  virtual void send_query_to_server(std::string name, td::BufferSlice data, size_t server_idx, td::Timestamp timeout,
+                                    td::Promise<td::BufferSlice> promise) {
+    promise.set_error(td::Status::Error("not supported"));
+  }
+  virtual void get_servers_status(td::Promise<std::vector<bool>> promise) {
+    promise.set_error(td::Status::Error("not supported"));
+  }
   virtual void reset_servers() {
   }
 
   static td::actor::ActorOwn<ExtClient> create(ton::adnl::AdnlNodeIdFull dst, td::IPAddress dst_addr,
                                                td::unique_ptr<Callback> callback);
   static td::actor::ActorOwn<ExtClient> create(std::vector<LiteServerConfig> liteservers,
-                                               td::unique_ptr<Callback> callback);
+                                               td::unique_ptr<Callback> callback, bool connect_to_all = false);
 };
 }  // namespace liteclient

lite-client/lite-client.cpp

@@ -3412,9 +3412,7 @@ void TestNode::got_creator_stats(ton::BlockIdExt req_blkid, ton::BlockIdExt blki
         promise.set_error(td::Status::Error(PSLICE() << "invalid CreatorStats record with key " << key.to_hex()));
         return;
       }
-      if (mc_cnt.modified_since(min_utime) || shard_cnt.modified_since(min_utime)) {
-        func(key, mc_cnt, shard_cnt);
-      }
+      func(key, mc_cnt, shard_cnt);
       allow_eq = false;
     }
     if (complete) {