replace slot_subscribe with get_slot

Author: jhunsaker

pc/manager.cpp

@@ -62,7 +62,8 @@ manager::manager()
   do_cap_( false ),
   do_tx_( true ),
   is_pub_( false ),
-  cmt_( commitment::e_confirmed )
+  cmt_( commitment::e_confirmed ),
+  sreq_{ { commitment::e_processed } }
 {
   tconn_.set_sub( this );
   breq_->set_sub( this );
@@ -433,6 +434,15 @@ void manager::poll( bool do_wait )
   // get current time
   curr_ts_ = get_now();
 
+  // get current slot
+  if ( curr_ts_ - slot_ts_ > 200 * PC_NSECS_IN_MSEC ) {
+    if ( sreq_->get_is_recv() ) {
+      if ( has_status( PC_PYTH_RPC_CONNECTED ) ) {
+        clnt_.send( sreq_ );
+      }
+    }
+  }
+
   // try to (re)connect to tx proxy
   if ( do_tx_ && ( !tconn_.get_is_connect() || tconn_.get_is_err() ) ) {
     tconn_.reconnect();
@@ -489,8 +499,9 @@ void manager::reconnect_rpc()
     kidx_ = 0;
     ctimeout_ = PC_NSECS_IN_SEC;
     pub_ts_ = 0L;
-    slot_cnt_ = 0UL;
     slot_ = 0L;
+    slot_cnt_ = 0UL;
+    slot_ts_ = 0L;
     num_sub_ = 0;
     clnt_.reset();
     for(;;) {
@@ -634,23 +645,30 @@ void manager::schedule( price_sched *kptr )
   }
 }
 
-void manager::on_response( rpc::slot_subscribe *res )
+void manager::on_response( rpc::get_slot *res )
 {
   // check error
   if ( PC_UNLIKELY( res->get_is_err() ) ) {
-    set_err_msg( "failed to slot_subscribe ["
+    set_err_msg( "failed to get slot ["
         + res->get_err_msg()  + "]" );
     return;
   }
 
   // ignore slots that go back in time
-  uint64_t slot = res->get_slot();
+  uint64_t slot = res->get_current_slot();
   int64_t ts = res->get_recv_time();
   if ( slot <= slot_ ) {
     return;
   }
   slot_ = slot;
-  PC_LOG_DBG( "receive slot" ).add( "slot", slot_ ).end();
+  slot_ts_ = ts;
+
+  int64_t ack_ts = res->get_recv_time() - res->get_sent_time();
+
+  PC_LOG_DBG( "received get_slot" )
+    .add( "slot", slot_ )
+    .add( "rount_trip_time(ms)", 1e-6*ack_ts )
+    .end();
 
   // submit block hash every N slots
   if ( slot_cnt_++ % PC_BLOCKHASH_TIMEOUT == 0 ) {

pc/manager.hpp

@@ -48,7 +48,7 @@ namespace pc
                   public net_accept,
                   public tx_sub,
                   public rpc_sub,
-                  public rpc_sub_i<rpc::slot_subscribe>,
+                  public rpc_sub_i<rpc::get_slot>,
                   public rpc_sub_i<rpc::get_recent_block_hash>,
                   public rpc_sub_i<rpc::program_subscribe>
   {
@@ -164,7 +164,7 @@ namespace pc
     bool get_is_tx_send() const;
 
     // rpc callbacks
-    void on_response( rpc::slot_subscribe * ) override;
+    void on_response( rpc::get_slot * ) override;
     void on_response( rpc::get_recent_block_hash * ) override;
     void on_response( rpc::program_subscribe * ) override;
     void set_status( int );
@@ -228,6 +228,7 @@ namespace pc
     int64_t      ctimeout_; // connection timeout
     uint64_t     slot_;     // current slot
     uint64_t     slot_cnt_; // slot count
+    int64_t      slot_ts_;  // current slot time
     int64_t      curr_ts_;  // current time
     int64_t      pub_ts_;   // start publish time
     int64_t      pub_int_;  // publish interval
@@ -241,7 +242,7 @@ namespace pc
     commitment   cmt_;      // account get/subscribe commitment
 
     // requests
-    rpc::slot_subscribe        sreq_[1]; // slot subscription
+    rpc::get_slot              sreq_[1]; // slot subscription
     rpc::get_recent_block_hash breq_[1]; // block hash request
     rpc::program_subscribe     preq_[1]; // program account subscription
   };

pc/rpc_client.cpp

@@ -637,6 +637,38 @@ void rpc::get_cluster_nodes::response( const jtree& jt )
   on_response( this );
 }
 
+///////////////////////////////////////////////////////////////////////////
+// get_slot
+
+rpc::get_slot::get_slot( commitment const cmt )
+: cmt_{ cmt }
+, cslot_( 0UL )
+{
+}
+
+uint64_t rpc::get_slot::get_current_slot() const
+{
+  return cslot_;
+}
+
+void rpc::get_slot::request( json_wtr& msg )
+{
+  msg.add_key( "method", "getSlot" );
+  msg.add_key( "params", json_wtr::e_arr );
+  msg.add_val( json_wtr::e_obj );
+  msg.add_key( "commitment", commitment_to_str( cmt_ ) );
+  msg.pop();
+  msg.pop();
+}
+
+void rpc::get_slot::response( const jtree& jt )
+{
+  if ( on_error( jt, this ) ) return;
+  uint32_t rtok = jt.find_val( 1, "result" );
+  cslot_ = jt.get_uint( rtok );
+  on_response( this );
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // get_slot_leaders
 

pc/rpc_client.hpp

@@ -366,6 +366,19 @@ namespace pc
       node_map_t nmap_;
     };
 
+    // get current slot
+    class get_slot : public rpc_request
+    {
+    public:
+      get_slot( commitment = e_finalized );
+      uint64_t get_current_slot() const;
+      void request( json_wtr& ) override;
+      void response( const jtree& ) override;
+    private:
+      commitment const cmt_; // param
+      uint64_t cslot_; // result
+    };
+
     // get id of leader node by slot
     class get_slot_leaders : public rpc_request
     {
@@ -376,7 +389,7 @@ namespace pc
       pub_key *get_leader( uint64_t );
       uint64_t get_last_slot() const;
       void request( json_wtr& ) override;
-      void response( const jtree&p) override;
+      void response( const jtree& ) override;
     private:
       typedef std::vector<pub_key> ldr_vec_t;
       uint64_t  rslot_;

pctest/slots_info.cpp

@@ -13,7 +13,7 @@ class manager_slot : public manager,
 {
 public:
   manager_slot();
-  void on_response( rpc::slot_subscribe * ) override;
+  void on_response( rpc::get_slot * ) override;
   void on_response( rpc::get_slot_leaders * ) override;
 private:
   rpc::get_slot_leaders ldr_[1];
@@ -27,10 +27,10 @@ manager_slot::manager_slot()
   ldr_->set_limit( PC_LEADER_MAX );
 }
 
-void manager_slot::on_response( rpc::slot_subscribe *res )
+void manager_slot::on_response( rpc::get_slot *res )
 {
   manager::on_response( res );
-  uint64_t slot = get_slot();
+  uint64_t const slot = get_slot();
   if ( slot != last_ ) {
     // request next slot leader schedule
     if ( PC_UNLIKELY( ldr_->get_is_recv() &&