Simplify notify_price_sched logic.

This commit updates the price_sched logic to schedule all price_sched
requests in both managers at a fixed interval. The reason for this is:

Previously, the notify_price_sched messages were staggered over the slot
interval. However, as the updates are now batched there is no need to
do this. Therefore, we now send notify_price_sched messages at a fixed
interval.

When subscribe_price_sched messages come in, the associated price_sched
requests are added to the manager which contains the price information.
This is necessary, as the manager associated with the price_sched
request needs to be the same manager as that of the price object.
This means that the price_sched requests the user cares about could be
distributed between both managers.

Therefore, at any time, all price_sched requests in either manager need
to be sent at a constant interval, regardless of their connectivity
status. This ensures that price updates are always being submitted to
pythd. If a network isn't online, its price updates will simply fail to
be sent.

Author: tompntn

pc/manager.cpp

@@ -60,20 +60,17 @@ manager::manager()
   sub_( nullptr ),
   status_( 0 ),
   num_sub_( 0 ),
-  kidx_( (unsigned)-1 ),
   cts_( 0L ),
   ctimeout_( PC_NSECS_IN_SEC ),
   slot_( 0UL ),
   slot_cnt_( 0UL ),
   slot_ts_{ 0UL },
   curr_ts_( 0L ),
-  pub_ts_( 0L ),
   pub_int_( PC_PUB_INTERVAL ),
   wait_conn_( false ),
   do_cap_( false ),
   do_ws_( true ),
   do_tx_( true ),
-  is_pub_( false ),
   cmt_( commitment::e_confirmed ),
   max_batch_( PC_MAX_BATCH ),
   sreq_{ { commitment::e_processed } },
@@ -572,6 +569,9 @@ void manager::poll( bool do_wait )
     }
   }
 
+  // request product quotes from publishers
+  poll_schedule();
+
   // try to (re)connect to tx proxy
   if ( do_tx_ && ( !tconn_.get_is_connect() || tconn_.get_is_err() ) ) {
     tconn_.reconnect();
@@ -580,9 +580,6 @@ void manager::poll( bool do_wait )
   if ( has_status( PC_PYTH_RPC_CONNECTED ) &&
        !hconn_.get_is_err() &&
        ( !wconn_ || !wconn_->get_is_err() ) ) {
-    // request product quotes from pythd's clients while connected
-    poll_schedule();
-
     send_pending_ups();
   } else {
     reconnect_rpc();
@@ -596,19 +593,10 @@ void manager::poll( bool do_wait )
 
 void manager::poll_schedule()
 {
-  while ( is_pub_ && kidx_ < kvec_.size() ) {
-    price_sched *kptr = kvec_[kidx_];
-    int64_t pub_ts = pub_ts_ + static_cast< int64_t >(
-      ( static_cast< uint64_t >( pub_int_ ) * kptr->get_hash() )
-        / price_sched::fraction
-    );
-    if ( curr_ts_ > pub_ts ) {
+  for ( uint32_t i = 0; i < kvec_.size(); i++ ) {
+    price_sched *kptr = kvec_[i];
+    if ( ( curr_ts_ - kptr->get_last_scheduled_time() ) > pub_int_ ) {
       kptr->schedule();
-      if ( ++kidx_ >= kvec_.size() ) {
-        is_pub_ = false;
-      }
-    } else {
-      break;
     }
   }
 }
@@ -633,10 +621,7 @@ void manager::reconnect_rpc()
 
     // reset state
     wait_conn_ = false;
-    is_pub_ = false;
-    kidx_ = 0;
     ctimeout_ = PC_NSECS_IN_SEC;
-    pub_ts_ = 0L;
     slot_ = 0L;
     slot_cnt_ = 0UL;
     slot_ts_ = 0L;
@@ -830,13 +815,6 @@ void manager::on_response( rpc::get_slot *res )
     clnt_.send( breq_ );
   }
 
-  // reset submit
-  if ( !is_pub_ ) {
-    kidx_ = 0;
-    pub_ts_ = ts;
-    is_pub_ = true;
-  }
-
   // flush capture
   if ( do_cap_ ) {
     cap_.flush();

pc/manager.hpp

@@ -255,14 +255,12 @@ namespace pc
     manager_sub *sub_;      // subscription callback
     int          status_;   // status bitmap
     int          num_sub_;  // number of in-flight mapping subscriptions
-    uint32_t     kidx_;     // schedule index
     int64_t      cts_;      // (re)connect timestamp
     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
     kpx_vec_t    kvec_;     // symbol price scheduling
     bool         wait_conn_;// waiting on connection

pc/request.cpp

@@ -1084,7 +1084,8 @@ void price::dump_json( json_wtr& wtr ) const
 
 price_sched::price_sched( price *ptr )
 : ptr_( ptr ),
-  shash_( 0UL )
+  shash_( 0UL ),
+  last_scheduled_time_( 0L )
 {
 }
 
@@ -1101,9 +1102,25 @@ uint64_t price_sched::get_hash() const
 bool price_sched::get_is_ready()
 {
   manager *cptr = get_manager();
-  return cptr->has_status( PC_PYTH_RPC_CONNECTED |
-                           PC_PYTH_HAS_BLOCK_HASH |
-                           PC_PYTH_HAS_MAPPING );
+
+  bool primary_ready = cptr->has_status(
+    PC_PYTH_RPC_CONNECTED |
+    PC_PYTH_HAS_BLOCK_HASH |
+    PC_PYTH_HAS_MAPPING
+  );
+
+  bool secondary_ready = cptr->has_secondary() &&
+    cptr->get_secondary()->has_status( 
+        PC_PYTH_RPC_CONNECTED |
+        PC_PYTH_HAS_BLOCK_HASH |
+        PC_PYTH_HAS_MAPPING
+    );
+
+  return primary_ready || secondary_ready;
+}
+
+int64_t price_sched::get_last_scheduled_time() {
+  return last_scheduled_time_;
 }
 
 void price_sched::submit()
@@ -1116,6 +1133,7 @@ void price_sched::submit()
 
 void price_sched::schedule()
 {
+  last_scheduled_time_ = get_now();
   on_response_sub( this );
 }
 

pc/request.hpp

@@ -203,6 +203,9 @@ namespace pc
     // get associated symbol price
     price *get_price() const;
 
+    // last time this price submision schedule was scheduled 
+    int64_t get_last_scheduled_time();
+
   public:
     static const uint64_t fraction = 997UL;
 
@@ -214,6 +217,7 @@ namespace pc
   private:
     price   *ptr_;
     uint64_t shash_;
+    int64_t last_scheduled_time_;
   };
 
   // price account re-initialized

pc/user.cpp

@@ -274,6 +274,10 @@ void user::parse_sub_price_sched( uint32_t tok, uint32_t itok )
     pub_key pkey;
     pkey.init_from_text( jp_.get_str( ntok ) );
     price *sptr = sptr_->get_price( pkey );
+    if ( PC_UNLIKELY( !sptr && sptr_->has_secondary() ) ) {
+      sptr = sptr_->get_secondary()->get_price( pkey );
+    }
+
     if ( PC_UNLIKELY( !sptr ) ) { add_unknown_symbol(itok); return; }
 
     // add subscription